连接单个数据库

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来创建这个库下的所有表了

而且多数据库时注意循环依赖问题