堆栈

我正在使用：

• Python 3.10.x
• 快速API 0.75.x
• SQLAlchemy 1.4.3x版

摘要

我正在为几个遗留数据库构建一个统一的FastAPI项目（存储在MariaDB 10.3的后端-必须为一些遗留软件保留结构）。
我的SQLA安装程序使用数据库模块来执行以下操作：

/databases.py

import dotenv
import os

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

import .models as models

dotenv.load_dotenv()

engines = {
    'parts': create_engine("mysql+pymysql://" + os.environ['DB_URL'] + "/parts", pool_pre_ping=True, pool_recycle=300),
    'shop': create_engine("mysql+pymysql://" + os.environ['DB_URL'] + "/shop", pool_pre_ping=True, pool_recycle=300),
    'purchasing': create_engine("mysql+pymysql://" + os.environ['DB_URL'] + "/purchasing", pool_pre_ping=True, pool_recycle=300),
    "company": create_engine("mysql+pymysql://" + os.environ['DB_URL'] + "/company", pool_pre_ping=True, pool_recycle=300),
    "auth": create_engine("mysql+pymysql://" + os.environ['DB_URL'] + "/auth", pool_pre_ping=True, pool_recycle=300),
}

DBSession = sessionmaker(autocommit=False, autoflush=False, binds={
    # Catalogue
    models.Shop.Catalogue: engines["shop"],
    models.Shop.Sections: engines["shop"],
    models.Shop.Orders: engines["shop"],
    # ...
    # Parts
    models.Parts.Part: engines["parts"],
    models.Parts.BinLocations: engines["parts"],

    # ...
    #Purchasing
    models.Purchasing.SupplierOrder: engines["purchasing"],
    models.Purchasing.SupplierOrder: engines["purchasing"],
    # Company Data
    models.Company.Staffmember: engines["company"],
    models.Company.Suppliers: engines["company"],
    # API Auth
    models.Auth.User: engines["auth"],
    models.Auth.Privileges: engines["auth"],
})

# Dependency

def getDb():
    db = DBSession()
    try:
        yield db
    finally:
        db.close()

对每个模型都这样做有点费力，但它确实有效。
由于我有几个数据库，我认为为每个数据库创建一个带有子文件的models模块是合乎逻辑的，例如models.Parts、models.Shop、models.Purchase、models.Company、models.Auth等。
/模型/初始化.py

from importlib.metadata import metadata
from sqlalchemy.orm import declarative_base

base = declarative_base()

from . import Auth, Parts, Shop, Catalogue, Purchasing, Shop

我可以通过将Base对象导入models的__init__.py并将其导入到每个子文件中来成功创建关系。例如：

/型号/Auth.py

from . import base as Base

from sqlalchemy.orm import relationship
from sqlalchemy import Column, Integer, String, Numeric, Date, DateTime, ForeignKey, null, or_, and_

class User(Base):
    __tablename__ = 'users'

    id = Column(Integer, nullable=False, primary_key=True)
    username = Column(String(256), nullable=False)
    passhash = Column(String(512), nullable=False)
    email = Column(String, nullable=False)
    enabled = Column(Integer, nullable=True)
    staffmember_id = Column(Integer, nullable=False)

    staffmember = relationship("Company.Staffmember", uselist=False)

/型号/Company.py

from . import base as Base

from sqlalchemy.orm import relationship
from sqlalchemy import Column, Integer, String, Numeric, Date, DateTime, ForeignKey, null, or_, and_

class Staffmebmer(Base):
    __tablename__ = 'staffmembers'

    id = Column(Integer, ForeignKey("users.staffmember_id"), nullable=False, primary_key=True)
    order = Column(Integer, default=0, nullable=False)
    name = Column(String, nullable=True)
    initial = Column(String, nullable=True)
    email = Column(String, nullable=False)
    enabled = Column(Integer, default=0, nullable=False)

    relationship("Auth.User", back_populates="staffmember")

下面的方法效果很好：
第1001章：我的demo.py

from fastapi import Depends

from sqlalchemy.orm import Session

from .. import app, databases, models

@app.get("/api/user/{id}")
async def read_items(id: int, db: Session=Depends(databases.getDb)):
    user = db.query(models.Auth.User).filter(
        models.Auth.User.id == id
    ).first()

    user.staffmember

    return user

访问此URL将返回：（是的，我知道这不安全，它仅用于说明目的，以显示关系的功能！）

{
  "username": "mark",
  "passhash": "<my hash>",
  "enabled": 1,
  "email": "mark@demo.com",
  "id": 1,
  "staffmember_id": 5,
  "staffmember": {
    "order": 20,
    "name": "Mark",
    "email": "mark@demo.com",
    "kStaffmember": 5,
    "initial": "MB",
    "enabled": 1
  }
}

但是，我想使用steffmember的首字母作为可能的用户名，所以当我在OAUTH Authorize脚本中查询用户时，我尝试使用：

from ..models import Auth, Company

# 'username' is provided by the auth script from the standard username/password OAuth fields

def get_user(db: Session, username: str):
    db_user_data = db.query(Auth.User).join(Company.Staffmember).filter(
        or_(
            Auth.User.username == username,
            Auth.User.email == username,
            Company.Staffmember.initial == username
        )
    ).first()

我得到一个异常：

(pymysql.err.ProgrammingError) (1146, "Table 'auth.staffmembers' doesn't exist")

我是否以正确的方式处理这整件事？有没有可能解决这个问题？