Блог

Можем ли мы объявить несколько сеансов баз данных в fast api таким образом?

 
Base = declarative_base()

@contextmanager
def get_db(db_type: str):
    if db_type == "retention":
        db = RetentionSessionLocal()
    else:
        db = SessionLocal()
    try:
        yield db
        db.commit()
    except Exception:
        db.rollback()
        raise
    finally:
        db.close()
  

Где RetentionSessionLocal указывает на другую базу данных, а SessionLocal указывает на основную БД.

А также является ли declarative_base() одинаковым для любой базы данных или его нужно изменить?

И если вышесказанное можно сделать, то как мы можем объявлять модели в соответствующих базах данных для выполнения операций CRUD?

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

 engine = create_engine(DATABASE_URL, pool_pre_ping=True)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

RETENTION_DATABASE_URL = "postgresql psycopg2://{0}:{1}@{2}:{3}/{4}".format(
    os.environ["DB_USERNAME"],
    os.environ["DB_PASSWORD"],
    os.environ["DB_HOST"], "5432",
    os.environ["RETENTION_DB"]
)
retention_engine = create_engine(RETENTION_DATABASE_URL, pool_pre_ping=True)
RetentionSessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=retention_engine)
MainBase = declarative_base()
RetentionBase = declarative_base()