安装时自动安装Pydantic, SQLAlchemy
示例代码：
from typing import List, Optional
from sqlmodel import Field, Session, SQLModel, create_engine, select,Relationship
"""
设置数据库参数
"""
MYSQL_DB = 'xhaccount' #数据库命
MYSQL_USER = 'root' #数据库账号
MYSQL_PASSWD = '123456' #数据库登陆密码
MYSQL_HOST = '127.0.0.1' #数据库地址
MYSQL_POST = 3306 #端口
basedir = os.path.abspath(os.path.dirname(__file__)) #路径
SQLALCHEMY_DATABASE_URI = "mysql://%s:%s@%s:%s/%s?charset=utf8" %(
MYSQL_USER, MYSQL_PASSWD, MYSQL_HOST, MYSQL_POST, MYSQL_DB) # 数据库连接url
SQLALCHEMY_MIGRATE_REPO = os.path.join(basedir, 'db_repository') # 设置数据库迁移保存的文件夹，用来sqlalchemymigrate
class Team(SQLModel, table=True):
"""Team表"""
id: Optional[int] = Field(default=None, primary_key=True) #主键
name: str
headquarters: str
heroes: List["Hero"] = Relationship(back_populates="team") #关联对象
class Hero(SQLModel, table=True):
"""Hero表"""
id: Optional[int] = Field(default=None, primary_key=True) #主键
name: str
secret_name: str
age: Optional[int] = None
team_id: Optional[int] = Field(default=None, foreign_key="team.id") #外键
team: Optional[Team] = Relationship(back_populates="heroes") #关联对象
#SQLite不能设置连接池参数
engine = create_engine(SQLALCHEMY_DATABASE_URI, echo=True)
def create_db_and_tables():
#创建表
SQLModel.metadata.create_all(engine)
# Code above omitted ??
def create_heroes():
with Session(engine) as session:
team_preventers = Team(name="Preventers", headquarters="Sharp Tower")
team_z_force = Team(name="Z-Force", headquarters="Sister Margaret’s Bar")
session.add(team_preventers) #增加表记录
session.add(team_z_force)
session.commit() #提交
hero_deadpond = Hero(
name="Deadpond", secret_name="Dive Wilson", team_id=team_z_force.id
)
hero_rusty_man = Hero(
name="Rusty-Man",
secret_name="Tommy Sharp",
age=48,
team_id=team_preventers.id,
)
hero_spider_boy = Hero(name="Spider-Boy", secret_name="Pedro Parqueador")
session.add(hero_deadpond)
session.add(hero_rusty_man)
session.add(hero_spider_boy)
session.commit()
session.refresh(hero_deadpond)
session.refresh(hero_rusty_man)
session.refresh(hero_spider_boy)
print("Created hero:", hero_deadpond)
print("Created hero:", hero_rusty_man)
print("Created hero:", hero_spider_boy)
def create_heroes_relationships():
with Session(engine) as session:
team_preventers = Team(name="Preventers", headquarters="Sharp Tower")
team_z_force = Team(name="Z-Force", headquarters="Sister Margaret’s Bar")
hero_deadpond = Hero(
name="Deadpond", secret_name="Dive Wilson", team=team_z_force
)
hero_rusty_man = Hero(
name="Rusty-Man", secret_name="Tommy Sharp", age=48, team=team_preventers
)
hero_spider_boy = Hero(name="Spider-Boy", secret_name="Pedro Parqueador")
session.add(hero_deadpond)
session.add(hero_rusty_man)
session.add(hero_spider_boy)
session.commit()
session.refresh(hero_deadpond)
session.refresh(hero_rusty_man)
session.refresh(hero_spider_boy)
print("Created hero:", hero_deadpond)
print("Created hero:", hero_rusty_man)
print("Created hero:", hero_spider_boy)
hero_spider_boy.team = team_preventers
session.add(hero_spider_boy)
session.commit()
def update_heroes():
#更新
with Session(engine) as session:
statement = select(Hero).where(Hero.name == "Spider-Boy")
results = session.exec(statement)
hero = results.one()
print("Hero:", hero)
hero.age = 16
session.add(hero)
session.commit()
session.refresh(hero)
print("Updated hero:", hero)
def delete_heroes():
#删除
with Session(engine) as session:
statement = select(Hero).where(Hero.name == "Spider-Youngster")
results = session.exec(statement)
hero = results.one()
print("Hero: ", hero)
session.delete(hero)
def select_heroes():
#查询(单表)
with Session(engine) as session:
#hero = session.get(Hero,9001) #查询主键id值为9001的数据，返回为None
statement = select(Hero).offset(6).limit(3) #从第六行开始，最多显示三行
# statement = select(Hero.name).where(Hero.age >= 35).where(Hero.age < 40) #条件过滤并且只显示age字段 这里等于用==符号
# statement = select(Hero).where(Hero.age >= 35, Hero.age < 40)#条件过滤并且只显示age字段
results = session.exec(statement)
# print("Hero",results.first()) #显示第一条数据 或者使用results.one()
for hero in results:
print(hero)
def select_heroes_mulittable():
#查询(多表)
with Session(engine) as session:
statement = select(Hero, Team).where(Hero.team_id == Team.id)
results = session.exec(statement)
for hero, team in results:
print("Hero:", hero, "Team:", team)
def create_heroes_useteam():
#修改表数据(通过其他表数据)
with Session(engine) as session:
team_preventers = Team(name="Preventers", headquarters="Sharp Tower")
team_z_force = Team(name="Z-Force", headquarters="Sister Margaret’s Bar")
session.add(team_preventers)
session.add(team_z_force)
session.commit()
hero_deadpond = Hero(
name="Deadpond", secret_name="Dive Wilson", team_id=team_z_force.id
)
hero_rusty_man = Hero(
name="Rusty-Man",
secret_name="Tommy Sharp",
age=48,
team_id=team_preventers.id,
)
hero_spider_boy = Hero(name="Spider-Boy", secret_name="Pedro Parqueador")
session.add(hero_deadpond)
session.add(hero_rusty_man)
session.add(hero_spider_boy)
session.commit()
session.refresh(hero_deadpond)
session.refresh(hero_rusty_man)
session.refresh(hero_spider_boy)
print("Created hero:", hero_deadpond)
print("Created hero:", hero_rusty_man)
print("Created hero:", hero_spider_boy)
hero_spider_boy.team_id = team_preventers.id
session.add(hero_spider_boy)
session.commit()
session.refresh(hero_spider_boy)
print("Updated hero:", hero_spider_boy)
def create_heros_read_relationships():
with Session(engine) as session:
team_preventers = Team(name="Preventers", headquarters="Sharp Tower")
team_z_force = Team(name="Z-Force", headquarters="Sister Margaret’s Bar")
hero_deadpond = Hero(
name="Deadpond", secret_name="Dive Wilson", team=team_z_force
)
hero_rusty_man = Hero(
name="Rusty-Man", secret_name="Tommy Sharp", age=48, team=team_preventers
)
hero_spider_boy = Hero(name="Spider-Boy", secret_name="Pedro Parqueador")
session.add(hero_deadpond)
session.add(hero_rusty_man)
session.add(hero_spider_boy)
session.commit()
session.refresh(hero_deadpond)
session.refresh(hero_rusty_man)
session.refresh(hero_spider_boy)
print("Created hero:", hero_deadpond)
print("Created hero:", hero_rusty_man)
print("Created hero:", hero_spider_boy)
hero_spider_boy.team = team_preventers
session.add(hero_spider_boy)
session.commit()
session.refresh(hero_spider_boy)
print("Updated hero:", hero_spider_boy)
hero_black_lion = Hero(name="Black Lion", secret_name="Trevor Challa", age=35)
hero_sure_e = Hero(name="Princess Sure-E", secret_name="Sure-E")
team_wakaland = Team(
name="Wakaland",
headquarters="Wakaland Capital City",
heroes=[hero_black_lion, hero_sure_e],
)
session.add(team_wakaland)
session.commit()
session.refresh(team_wakaland)
print("Team Wakaland:", team_wakaland)
#一次性创建一行
hero_tarantula = Hero(name="Tarantula", secret_name="Natalia Roman-on", age=32)
hero_dr_weird = Hero(name="Dr. Weird", secret_name="Steve Weird", age=36)
hero_cap = Hero(
name="Captain North America", secret_name="Esteban Rogelios", age=93
)
#分别设置各行
team_preventers.heroes.append(hero_tarantula)
team_preventers.heroes.append(hero_dr_weird)
team_preventers.heroes.append(hero_cap)
#增加
session.add(team_preventers)
#提交
session.commit()
#刷新表及缓存
session.refresh(hero_tarantula)
session.refresh(hero_dr_weird)
session.refresh(hero_cap)
print("Preventers new hero:", hero_tarantula)
print("Preventers new hero:", hero_dr_weird)
print("Preventers new hero:", hero_cap)
def select_heros_read_relationships():
with Session(engine) as session:
#查询
statement = select(Hero).where(Hero.name == "Spider-Boy")
result = session.exec(statement)
#获取查询结果(一条)，可以用first()来替代one()
hero_spider_boy = result.one()
statement = select(Team).where(Team.id == hero_spider_boy.id)
result = session.exec(statement)
#获取查询结果记录集里的第一条
team = result.first()
print("Spider-Boy's team:", team)
print("Spider-Boy's team again:", hero_spider_boy.team)