SQLAlchemy: ПРИСОЕДИНЯЙТЕСЬ к разным базам данных И используйте разные файлы в модуле

Куча

Я использую:

• Питон 3.10.x
• FastAPI 0.75.x
• SQLAlchemy 1.4.3x

Резюме

Я создаю объединяющий проект FastAPI для нескольких устаревших баз данных (хранится на базе MariaDB 10.3 — структура должна быть сохранена для некоторого устаревшего программного обеспечения).

Моя установка SQLA использует модуль баз данных, чтобы сделать следующее:

/базы данных.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 в __init__.py из models и импортируя его в каждый подфайл. Например:

/модели/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")

Следующий маршрут работает нормально:

демо.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": "[email protected]",
  "id": 1,
  "staffmember_id": 5,
  "staffmember": {
    "order": 20,
    "name": "Mark",
    "email": "[email protected]",
    "kStaffmember": 5,
    "initial": "MB",
    "enabled": 1
  }
}

Однако я хочу использовать инициалы steffmember в качестве возможного имени пользователя, поэтому, когда я запрашивал пользователя в своих сценариях авторизации OAUTH, я пытался использовать:

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

Правильно ли я делаю все это, и есть ли возможный способ обойти эту проблему?