Using two SQLite database connection in-memory.
from sqlalchemy import create_engine, MetaData
from sqlalchemy import text
from sqlalchemy.orm import DeclarativeBase
from sqlalchemy import select
from sqlalchemy import insert
from sqlalchemy.orm import Mapped, mapped_column
engine_1 = create_engine("sqlite:///:memory:", echo=False)
engine_2 = create_engine("sqlite:///:memory:", echo=False)
class Base1(DeclarativeBase):
metadata = MetaData(schema="blog")
class Base2(DeclarativeBase):
metadata = MetaData(schema="user")
with engine_1.connect() as conn:
conn.execute(text("attach database 'file:/memdb1?vfs=memdb' as blog;"))
with engine_2.connect() as conn:
conn.execute(text("attach database 'file:/memdb1?vfs=memdb' as user;"))
class News(Base1):
__tablename__ = "news"
id: Mapped[int] = mapped_column(primary_key=True)
title: Mapped[str] = mapped_column()
class Role(Base2):
__tablename__ = "role"
id: Mapped[int] = mapped_column(primary_key=True)
name: Mapped[str] = mapped_column()
Base1.metadata.create_all(engine_1)
Base2.metadata.create_all(engine_2)
with engine_1.connect() as conn:
stmt = insert(News).values(title="Python Energy")
conn.execute(stmt)
conn.commit()
print("----------------")
stmt = select(News)
for row in conn.execute(stmt):
print(row)
with engine_2.connect() as conn:
stmt = insert(Role).values(name="John Doe")
conn.execute(stmt)
conn.commit()
print("----------------")
stmt = select(Role)
for row in conn.execute(stmt):
print(row)