SQLAlchemy

代码片断

通过反射自动定义列

from sqlalchemy.ext.declarative import declarative_base, DeferredReflection

Base = declarative_base(cls=DeferredReflection)

class Star(Base):
  __tablename__ = 'star'
  # for MySQL to select database
  __table_args__ = {'schema': 'wodfan'}

def setup(engine):
  Base.prepare(engine)

Session 上下文管理器

Session = sessionmaker() # then call .configure(bind=engine) somewhere

@contextmanager
def get_session():
  session = Session()
  try:
    yield session
    session.commit()
  except:
    session.rollback()
    raise
  finally:
    session.close()

在连接时执行操作

使用 PoolListener ，如为奇葩的 MySQL 设置一个比较现代的字符集：

from sqlalchemy.interfaces import PoolListener

class UseRealUTF8(PoolListener):
  def connect(self, dbapi_con, con_record):
    dbapi_con.cursor().execute('set names utf8mb4;')

# pool_recycle is to make MySQL not go away
engine = sqlalchemy.create_engine(db_url, pool_recycle=3600, listeners=[UseRealUTF8()])

使用 MySQL-python 时，指定使用 utf8mb4 编码会报错。

关闭当前连接

有时时候需要关闭当前连接，如当前连接通过 fork 发生共享时：

session.bind.dispose()

session.get_bind() 有时候不好用。

注意事项

DateTime 类型默认是不带时区的

DateTime 类型在 PostgreSQL 中会生成 timestamp without time zone 类型。要生成带时区的类型，使用 TIMESTAMP(timezone=True) 。

参见

• psycopg2
• MySQL Connector/Python

外部链接

• The Architecture of Open Source Applications (Volume 2): SQLAlchemy