官网总教程

所有代码基于sqlalchemy,官网有极其详细的英文教程

创建表

sqlalchemy官网教程:Column的写法
sqlalchemy官网教程:用ORM方式创建表

  1. from sqlalchemy import create_engine
  2. from sqlalchemy import Column, Integer,String, DateTime
  3. from sqlalchemy.orm import declarative_base
  4. Base = declarative_base()
  5. class Test1(Base):
  6. __tablename__ = "test1"
  7. id = Column(Integer, primary_key=True)
  8. name = Column(String, nullable=False)
  9. lastuse = Column("上次使用", DateTime)
  10. hash = Column(String(16), nullable=False)
  11. engine = create_engine("sqlite:///aa.sqlite3", echo=True)
  12. Test1.metadata.create_all(engine)

写入新记录

python官网:日期时间的模式字符串
pandas官网:sql相关快捷函数
pandas函数to_sql
pandas函数to_dict
sqlalchemy官网:批量insert

  1. from sqlalchemy import select, update, insert
  2. # 注意这里的日期格式是sqlite标准日期格式,时分秒后面可以再加微秒,如12:31:54.800689。微秒的模式符为%f
  3. data_init = pd.DataFrame(
  4. {
  5. "id": [4, 5, 6],
  6. "name": ["name1", "name2", "name3"],
  7. "上次使用": pd.to_datetime(
  8. ["2022-02-03 15:00:00", "2022-02-01 16:00:00", "2022-01-03 09:00:23"],
  9. format="%Y-%m-%d %H:%M:%S",
  10. ),
  11. }
  12. )
  13. #========
  14. data_init.to_sql("test1", engine, if_exists="append", index=False)
  15. # 或者
  16. with engine.connect() as cursor:
  17. cursor.execute(insert(Test1).values(data_init.to_dict("split")["data"]))

更新记录

注意需要先确保这些记录的ID在数据库中已存在。

  1. from sqlalchemy import select, update, insert
  2. df = df.to_dict("records")
  3. with engine.connect() as cursor:
  4. for line in df:
  5. cursor.execute(update(Test1).where(Test1.id == line["id"]).values(line))

删除记录

  1. from sqlalchemy import select, update, insert
  2. IDs = df["id"].to_list()
  3. with engine.connect() as cursor:
  4. for item in IDs:
  5. cursor.execute(delete(Test1).where(Test1.id == item))

读取数据

pandas官网:read_sql_table
pandas官网:read_sql_query
通过engine的输出可以看出read_sql_query()确实只调用了一句SELECT ... FROM ...,而read_sql_table()进行了更多的操作。因此从性能的角度优选read_sql_query()

  1. df = pd.read_sql_table("test1", engine, columns=["id", "hash"])
  2. df = pd.read_sql_query(select(Test1.id,Test1.hash),engine)