以正确的方式构建 Flask 项目（无蓝图）

Question

简介

我在 SO 和其他网站上阅读了大约一千篇文章，试图找出我的 Flask 结构出了什么问题，以及为什么我似乎无法弄清楚某些事情。万不得已，我决定最后在这里问这个问题。

我的项目很简单：

1. 我必须通过 API 从一些网络设备获取一些数据，处理数据并将其存储在 Postgresql 数据库中（大部分代码在 lib/ 中）。
2. 该项目将部署在多个环境（测试、开发、登台和生产）上。

要执行上述操作，我使用以下内容：

• Flask-SQLAlchemy + Flask-Migrate + Flask-Script - 所有这些都用于处理迁移和数据库相关操作；

• Python-dotenv - 用于处理敏感配置数据

---

项目详情

我的项目结构如下：

my_project/
├── api/  
├── app.py             
├── config.py
├── __init__.py
├── lib/
│   ├── exceptions.py
│   └── f5_bigip.py
├── log.py
├── logs/
├── manage.py
├── migrations/
├── models/
│   ├── __init__.py
│   ├── model1.py
│   └── model2.py
└── run.py

我的 app.py 如下所示：

import os
import sys
sys.path.append(os.path.dirname(os.path.abspath(__file__)))

from dotenv import load_dotenv
from flask import Flask
from flask_migrate import Migrate
from flask_sqlalchemy import SQLAlchemy


db = SQLAlchemy()
migrate = Migrate()


def create_app():
    load_dotenv()

    app = Flask(__name__)

    environment = app.config['ENV']

    if environment == 'production':
        app.config.from_object('config.ProductionConfig')
    elif environment == 'testing':
        app.config.from_object('config.TestingConfig')
    else:
        app.config.from_object('config.DevelopmentConfig')

    db.init_app(app)
    migrate.init_app(app, db)

    return app

我的 config.py 如下所示：

import os

from sqlalchemy.engine.url import URL

PROJECT_ROOT = os.path.dirname(os.path.realpath(__file__))


class BaseConfig:
    DEBUG = False
    TESTING = False

    DB_DRIVERNAME = os.getenv('DB_DRIVERNAME')
    DB_HOST = os.getenv('DB_HOST')
    DB_PORT = os.getenv('DB_PORT')
    DB_NAME = os.getenv('DB_NAME')
    DB_USERNAME = os.getenv('DB_USERNAME')
    DB_PASSWORD = os.getenv('DB_PASSWORD')

    DB = {
        'drivername': DB_DRIVERNAME,
        'host': DB_HOST,
        'port': DB_PORT,
        'database': DB_NAME,
        'username': DB_USERNAME,
        'password': DB_PASSWORD,
    }

    SQLALCHEMY_DATABASE_URI = URL(**DB)
    SQLALCHEMY_TRACK_MODIFICATIONS = False


class DevelopmentConfig(BaseConfig):
    DEVELOPMENT = True
    DEBUG = True


class TestingConfig(BaseConfig):
    TESTING = True


class StagingConfig(BaseConfig):
    DEVELOPMENT = True
    DEBUG = True


class ProductionConfig(BaseConfig):
    pass

我的 __init__.py 看起来像这样：

from contextlib import contextmanager

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

# here, create_engine needs the SQLALCHEMY_DATABASE_URI
# how do I get it from the proper config?
engine = create_engine()
Session = sessionmaker(bind=engine)


@contextmanager
def session_scope():
    """
    Provide a transactional scope around a series of operations.
    """
    session = Session()
    try:
        yield session
        session.commit()
    except Exception as e:
        print(f'Something went wrong here: {str(e)}. rolling back.')
        session.rollback()
        raise
    finally:
        session.close()

我的 ma​​nage.py 看起来像这样：

from flask_script import Manager
from flask_migrate import MigrateCommand

from app import create_app


from models import *


manager = Manager(create_app)
manager.add_command('db', MigrateCommand)

if __name__ == '__main__':
    manager.run()

我的 models/model1.py 看起来像这样：

from sqlalchemy.dialects.postgresql import INET
from sqlalchemy.sql import func

from app import db


class Model1(db.Model):
    __tablename__ = 'model1'

    id = db.Column(db.BigInteger, primary_key=True, autoincrement=True)
    ip_address = db.Column(INET, unique=True, nullable=False)
    last_update = db.Column(db.DateTime(), server_default=func.now())

    def __repr__(self):
        return f'<Model1: {self.ip_address}>'

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

---

问题

现在，我有三个主要问题：

1. 在我的主要__init__.py 中，如何从应用的配置中导入SQLALCHEMY_DATABASE_URI？
2. 在__init__.py 中包含Session() 对象似乎不太直观。应该放在其他地方吗？有关更多上下文，session 用于lib/f5_bigip.py，并且可能也将用于api/。
3. 整个项目结构还好吗？

Answer 1

您的问题 1 和 2 与您的项目中我觉得奇怪的部分直接相关，因此我不会回答这些问题，而是给您一个更简单更好的方法。

在__init__.py 中，您似乎正在实现自己的数据库会话，以便您可以创建一个作用域会话上下文管理器。也许您从另一个项目中获得了该代码？它没有很好地与使用 Flask-SQLAlchemy 扩展的项目的其余部分集成，您只是忽略了 Flask-SQLAlchemy，它管理您的数据库连接和会话，并且基本上创建了另一个到数据库和新会话的连接。

您应该做的是利用 Flask-SQLAlchemy 提供的连接和会话。我会重写__init__.py 如下（凭记忆这样做，所以请原谅小错误）：

from contextlib import contextmanager

from app import db


@contextmanager
def session_scope():
    """
    Provide a transactional scope around a series of operations.
    """
    try:
        yield db.session
        session.commit()
    except Exception as e:
        print(f'Something went wrong here: {str(e)}. rolling back.')
        db.session.rollback()
        raise
    finally:
        db.session.close()

这样你就可以重用来自 Flask-SQLAlchemy 的连接/会话。那么你的问题 1 就不再是问题了。对于问题 2，您可以在应用中需要数据库会话的任何位置使用 db.session。

关于第 3 个问题，我认为您基本没问题。我建议你不要使用 Flask-Script，这是一个相当古老且无人维护的扩展。相反，您可以将 CLI 移至 Flask 自己的 CLI 支持。