官网总教程

所有代码基于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"
  8.     id = Column(Integer, primary_key=True)
  9.     name = Column(String, nullable=False)
  10.     lastuse = Column("上次使用", DateTime)
  11.     hash = Column(String(16), nullable=False)
  14. engine = create_engine("sqlite:///aa.sqlite3", echo=True)
  15. 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)