如何将 SQLAlchemy 相关代码拆分为模块

Question

我是 Python 新手。我正在使用 Flask 和 SQLAlchemy 创建一个 API。我正在寻找将 SQLAlchemy 代码拆分为模块的建议/解决方案。我正在尝试使用以下方法模块化我的代码：

• /models 将包含我的 SQLAlchemy 模型
• /methods 将包含各种数据库方法，例如创建、更新和删除用户
• /routes 将包含我的 Flask 路由处理程序

我引用了这个Stack Overflow question on how to split SQLAlchemy declarative models，它在一定程度上有效。使用下面显示的文件结构，我成功地将与数据库相关的代码拆分为所需的模块。

Directories are denoted with a leading '/'

/my_root_directory
|__ main.py                         Imports my Flask app and runs it
|__ /server
    |__ __init__.py                 Imports my Flask app and database modules
    |__ app.py                      Contains my Flask app
    |__ /database
        |__ __init__.py             Imports initialize_database()
        |__ engine.py               Contains SQLAlchemy's create_engine()
        |__ base.py                 Contains SQLAlchemy's declarative_base()
        |__ session.py              Contains SQLAlchemy's scoped_session(sessionmaker())
        |__ initialize_database.py  Contains initialize_database()
        |__ /models
            |__ __init__.py         Imports my SQLAlchemy User model
            |__ user.py             Contains my SQLAlchemy User model
        |__ /methods
            |__ __init__.py         Imports create_user()
            |__ create_user.py      Contains create_user()
    |__ /routes
        |__ __init__.py             Imports my Flask route handler
        |__ /api
            |__ users_router.py     Contains my Flask route handler for /api/v1/users

这是我的申请流程。我通过在终端中运行python main.py 来执行main.py，其中包含以下代码：

ma​​in.py

from server.app import app
import os

if __name__ == "__main__":
    debug = os.getenv("ENV") == "dev"
    app.run(debug=debug)

在 /server/__init__.py 中，我执行 (?) 我所有与数据库相关的代码：

/server/__init__.py

from .app import app
from .database import *

/server/database/__init__.py初始化数据库并测试我的数据库方法：

/server/database/__init__.py

from .initialize_database import initialize_database
from .methods.create_user import create_user

initialize_database()
print(create_user("jack"))
print(create_user("john"))

这成功地在我的数据库中创建了新用户。这是这两个函数的代码。为了完整起见，我还提供了包含Session和User的代码：

/server/database/initialize_database.py

from .engine import engine
from .base import Base
import os


def initialize_database():
    if os.getenv("ENV") == "dev":
        Base.metadata.drop_all(bind=engine)

    Base.metadata.create_all(bind=engine)

/server/database/methods/create_user.py

from ..session import Session
from ..models import User


def create_user(name):
    user = User(twitch_username=name)
    Session.add(user)
    Session.commit()
    return "Created new user"

/server/database/session.py

from sqlalchemy.orm import scoped_session, sessionmaker
from .engine import engine
Session = scoped_session(sessionmaker(bind=engine))

/server/database/models/user.py

from sqlalchemy import Column, String
from ..base import Base

# These SQLAlchemy model mixins don't depend on any relative modules
from .mixins import AutoUUIDMixin, TimeStampMixin


class User(Base, AutoUUIDMixin, TimeStampMixin):
    __tablename__ = "users"

    twitch_username = Column("twitch_username", String(50), unique=True)

    def __init__(self, twitch_username):
        self.twitch_username = twitch_username

---

但是，当我尝试通过执行以下操作将 create_user() 导入我的路由处理程序时：

/server/routes/api/users_router.py

from ...database.methods import create_user as CreateUser

我收到以下错误：

Traceback (most recent call last):
  File "main.py", line 1, in <module>
    from server.app import app
  File "~/root_dir/server/__init__.py", line 1, in <module>
    from .app import app
  File "~/root_dir/server/app.py", line 9, in <module>
    from .routes import users_router
  File "~/root_dir/server/routes/__init__.py", line 1, in <module>
    from .api import users_router
  File "~/root_dir/server/routes/api/__init__.py", line 1, in <module>
    from .users_router import users_router
  File "~/root_dir/server/routes/api/users_router.py", line 2, in <module>
    from ...database.methods import create_user as CreateUser
  File "~/root_dir/server/database/__init__.py", line 1, in <module>
    from .initialize_database import initialize_database
  File "~/root_dir/server/database/initialize_database.py", line 1, in <module>
    from .engine import engine
  File "~/root_dir/server/database/engine.py", line 3, in <module>
    engine = create_engine(os.getenv("DATABASE_URL"))
  File "/Library/Frameworks/Python.framework/Versions/3.7/lib/python3.7/site-packages/sqlalchemy/engine/__init__.py", line 479, in create_engine
    return strategy.create(*args, **kwargs)
  File "/Library/Frameworks/Python.framework/Versions/3.7/lib/python3.7/site-packages/sqlalchemy/engine/strategies.py", line 56, in create
    plugins = u._instantiate_plugins(kwargs)
AttributeError: 'NoneType' object has no attribute '_instantiate_plugins'

如何成功模块化我的数据库相关逻辑以供我的 Flask 应用程序使用？我对任何方法和解决方案持开放态度，即使这意味着完全重组我的代码库。我对 Python 模块和脚本没有最好的理解，所以如果可能的话，解释一下为什么我当前的文件结构没有按预期工作会很有帮助。

---

这是我的依赖项列表：

Python     3.7.4
Anaconda   4.8.3

requirements.txt

click==7.1.1
Flask==1.1.2
Flask-SQLAlchemy==2.4.1
gunicorn==20.0.4
itsdangerous==1.1.0
Jinja2==2.11.1
MarkupSafe==1.1.1
psycopg2==2.8.5
psycopg2-binary==2.8.5
python-dotenv==0.12.0
SQLAlchemy==1.3.15
Werkzeug==1.0.1

environment.yml

name: flask-server
channels:
  - defaults
dependencies:
  - ca-certificates=2020.1.1=0
  - libcxx=4.0.1=hcfea43d_1
  - libcxxabi=4.0.1=hcfea43d_1
  - libedit=3.1.20181209=hb402a30_0
  - libffi=3.2.1=h475c297_4
  - ncurses=6.2=h0a44026_0
  - openssl=1.1.1f=h1de35cc_0
  - pip=20.0.2=py37_1
  - python=3.7.6=h359304d_2
  - readline=7.0=h1de35cc_5
  - setuptools=46.1.3=py37_0
  - sqlite=3.31.1=ha441bb4_0
  - tk=8.6.8=ha441bb4_0
  - wheel=0.34.2=py37_0
  - xz=5.2.4=h1de35cc_4
  - zlib=1.2.11=h1de35cc_3
  - pip:
    - click==7.1.1
    - flask==1.1.2
    - flask-sqlalchemy==2.4.1
    - gunicorn==20.0.4
    - itsdangerous==1.1.0
    - jinja2==2.11.1
    - markupsafe==1.1.1
    - psycopg2==2.8.5
    - psycopg2-binary==2.8.5
    - python-dotenv==0.12.0
    - sqlalchemy==1.3.15
    - werkzeug==1.0.1

---

编辑：这是engine.py中的代码。

/server/database/engine.py

from sqlalchemy import create_engine
import os

print(os.getenv(“DATABASE_URL”))
# Sometimes prints None and other times prints “postgresql://localhost:5432/<my-db>“

engine = create_engine(os.getenv(“DATABASE_URL”))

Answer 1

在翻转代码执行后，我设法让我的文件结构正常工作。我没有先创建我的 Flask 应用程序，而是执行我的数据库逻辑。

/server/__init__.py

from .app import app
from .database import *

变成

from .database import *
from .app import app

我还在engine.py 中添加了load_dotenv()。

from sqlalchemy import create_engine
from dotenv import load_dotenv
import os

load_dotenv()

print(os.getenv("DATABASE_URL"))
# Prints “postgresql://localhost:5432/<my-db>“

engine = create_engine(os.getenv("DATABASE_URL"))