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

Question

简介

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

我的项目真的很简单：

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

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

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()

我的manage.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

问题

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

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

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 中的 connection/sessions。那么您的问题 1 就不再是问题了。对于问题 2，您可以在应用中需要数据库会话的任何地方使用 db.session。

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

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

Structuring your Flask project the right way (no Blueprint)

flask

configuration

flask-sqlalchemy

flask-migrate

flask-script

简介

项目详情

问题