python - 模块中的自定义引擎创建 SQLAlchemy

Question

我有一个项目，我想将数据库初始化(SQLAlchemy)与其他模块隔离开来，所以我创建了一个模块

db_initializer.py:

engine = create_engine('sqlite:///db')  # TODO from config file
Session = sessionmaker(bind=engine)
Base = declarative_base(bind=engine)


def create_tables():
    Base.metadata.create_all(engine)

首先我需要将 create_all 放在一个函数中，因为我的模型在另一个包中。

模型/foo.py:

from core.db_initializer import Base

class User(Base):
    __tablename__ = 'user'

    id = Column(Integer, primary_key=True)
    name = Column(String)

    def __init__(self, name: str = None):
        self.name = name

    def __repr__(self):
        return "<User(id=%d, name=%s)>" % (int(self.id), str(self.name))

还有我的主要调用 create_tables。

还有其他方法吗？现在我需要使用自定义配置(IP、用户、...)创建引擎，并且只有主脚本知道配置是否可行？

有点像

main.py:

import db_initializer as db
import model.foo
db.init(config) # which create the engine and create table 

当我做类似的事情时，我遇到了 Base 对象的问题，该对象尚未绑定(bind)到尚未创建的引擎。有什么解决办法吗？

Answer 1

您无需在声明模型之前创建引擎或 session 。您可以在 model/foo.py 中声明模型:

from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()

class User(Base):
    __tablename__ = 'user'

假设您在 myapp.py 中有一些应用程序。声明它以便它可以用引擎初始化:

from sqlalchemy.orm import Session
import model.foo

class MyApp:
    def __init__(self, engine):
        self.engine = engine

    def get_users(self):
        session = Session(self.engine)
        users = session.query(model.foo.User).all()
        session.close()

        return users

在 main.py 中创建引擎并使用它来初始化 models.foo.Base.metadata 和您需要它的其他应用程序:

from sqlalchemy import create_engine
import model.foo
import myapp

engine = create_engine('sqlite:///db')

model.foo.Base.metadata.bind = engine
model.foo.Base.metadata.create_all()

app = myapp.MyApp(engine)

UPD:对于 scoped_session 方法，myapp.py 可以如下所示:

import model.foo

class MyApp:
    def __init__(self, session):
        self.session = session

    def get_users(self):
        return self.session.query(model.foo.User).all()

和 main.py:

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, scoped_session
import model.foo
import myapp

engine = create_engine('sqlite:///db')
session = scoped_session(sessionmaker(engine))

model.foo.Base.metadata.bind = engine
model.foo.Base.metadata.create_all()

app = myapp.MyApp(session)