本文阅读量 18 次

1. SQLAlchemy 教程¶

1.1 表结构的定义¶

1.1.1 使用 Mixins 定义表结构¶

https://docs.sqlalchemy.org/en/20/orm/declarative_mixins.html ⧉

1.2 异步使用¶

同步风格异步风格

from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from sqlalchemy.orm import sessionmaker

engine = create_async_engine(<your_connection_string>, echo=True, future=True)

def async_session_generator():
    return sessionmaker(
        engine, class_=AsyncSession
    )

1.3 事件机制¶

1.3.1 注册事件的方式¶

注册函数 listen ⧉
装饰器方法 @listens_for ⧉

from sqlalchemy.event import listen
from sqlalchemy.pool import Pool


def my_on_connect(dbapi_con, connection_record):
    print("New DBAPI connection:", dbapi_con)


listen(Pool, "connect", my_on_connect)

@listens_for(Pool, "connect")
def my_on_connect(dbapi_con, connection_record):
    print("New DBAPI connection:", dbapi_con)

1.3.2 Event 分类¶

Event 可以分为 Core Events 和 ORM Events 两大类

1.4 参考¶