sqlalchemy连接多个数据库_异步形式
from fastapi import FastAPI
from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker, declarative_base
app = FastAPI()
engine1 = create_async_engine('sqlite+aiosqlite:///db1.sqlite3')
engine2 = create_async_engine('sqlite+aiosqlite:///db2.sqlite3')
engine3 = create_async_engine('sqlite+aiosqlite:///db3.sqlite3')
AsyncSession1 = sessionmaker(engine1, expire_on_commit=False, class_=AsyncSession)
AsyncSession2 = sessionmaker(engine2, expire_on_commit=False, class_=AsyncSession)
AsyncSession3 = sessionmaker(engine3, expire_on_commit=False, class_=AsyncSession)
Base1 = declarative_base()
Base2 = declarative_base()
Base3 = declarative_base()
class User(Base1):
__tablename__ = 'users'
id = Column(Integer, primary_key=True)
name = Column(String(64))
class Post(Base2):
__tablename__ = 'posts'
id = Column(Integer, primary_key=True)
title = Column(String(64))
class Comment(Base3):
__tablename__ = 'comments'
id = Column(Integer, primary_key=True)
body = Column(String(256))
async def create_tables():
async with engine1.begin() as conn:
await conn.run_sync(Base1.metadata.create_all)
async with engine2.begin() as conn:
await conn.run_sync(Base2.metadata.create_all)
async with engine3.begin() as conn:
await conn.run_sync(Base3.metadata.create_all)
@app.on_event("startup")
async def startup():
await create_tables()
@app.get('/')
async def index():
async with AsyncSession1() as session1, AsyncSession2() as session2, AsyncSession3() as session3:
users = await session1.execute(session1.query(User))
posts = await session2.execute(session2.query(Post))
comments = await session3.execute(session3.query(Comment))
return {'users': [user.name for user in users.scalars()],
'posts': [post.title for post in posts.scalars()],
'comments': [comment.body for comment in comments.scalars()]}
sqlalchemy连接多个数据库_同步形式
from fastapi import FastAPI
from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.orm import sessionmaker, declarative_base
app = FastAPI()
engine1 = create_engine('sqlite:///db1.sqlite3')
engine2 = create_engine('sqlite:///db2.sqlite3')
engine3 = create_engine('sqlite:///db3.sqlite3')
Session1 = sessionmaker(bind=engine1)
Session2 = sessionmaker(bind=engine2)
Session3 = sessionmaker(bind=engine3)
Base1 = declarative_base()
Base2 = declarative_base()
Base3 = declarative_base()
class User(Base1):
__tablename__ = 'users'
id = Column(Integer, primary_key=True)
name = Column(String(64))
class Post(Base2):
__tablename__ = 'posts'
id = Column(Integer, primary_key=True)
title = Column(String(64))
class Comment(Base3):
__tablename__ = 'comments'
id = Column(Integer, primary_key=True)
Base1.metadata.create_all(engine1)
Base2.metadata.create_all(engine2)
Base3.metadata.create_all(engine3)
@app.get('/')
def index():
session1 = Session1()
session2 = Session2()
session3 = Session3()
users = session1.query(User).all()
posts = session2.query(Post).all()
comments = session3.query(Comment).all()
return {'users': [user.name for user in users],
'posts': [post.title for post in posts],
'comments': [comment.body for comment in comments]}