连接单个数据库
from typing import Optional
from sqlmodel import Field, Session, SQLModel, create_engine
# 表
class Hero(SQLModel, table=True):
id: Optional[int] = Field(default=None, primary_key=True)
name: str
secret_name: str
age: Optional[int] = None
# 连接
engine = create_engine("sqlite:///database.db")
# 建表
SQLModel.metadata.create_all(engine)
连接多个数据库
from typing import Optional
from sqlmodel import Field, Session, SQLModel, create_engine
from sqlmodel.main import registry
# 表1
table1 = registry()
class Hero1(SQLModel, table=True, registry=table1):
id: Optional[int] = Field(default=None, primary_key=True)
name: str
secret_name: str
age: Optional[int] = None
# 表2
table2 = registry()
class Hero2(SQLModel, table=True, registry=table2):
id: Optional[int] = Field(default=None, primary_key=True)
name: str
secret_name: str
age: Optional[int] = None
# 连接1
engine1 = create_engine("sqlite:///database1.db")
# 连接2
engine2 = create_engine("sqlite:///database2.db")
# 建表1
Hero1.metadata.create_all(engine1)
# 建表2
Hero2.metadata.create_all(engine2)
一点优化
显然这样建多表会很麻烦,一种优化方案是:
class Base1(SQLModel, registry=registry()):
...
class Base2(SQLModel, registry=registry()):
...
engine1 = create_engine("sqlite:///database1.db")
engine2 = create_engine("sqlite:///database2.db")
class Hero1(Base1, table=True):
id: Optional[int] = Field(default=None, primary_key=True)
name: str
secret_name: str
age: Optional[int] = None
class Hero2(Base2, table=True):
id: Optional[int] = Field(default=None, primary_key=True)
name: str
secret_name: str
age: Optional[int] = None
# 建表1
Base1.metadata.create_all(engine1)
# 建表2
Base2.metadata.create_all(engine2)
此时无论有多少张表需要create都可以用基类的create_all来创建这个库下的所有表了
而且多数据库时注意循环依赖问题