安裝時自動安裝Pydantic, SQLAlchemy
示例代碼：
from typing import List, Optional
from sqlmodel import Field, Session, SQLModel, create_engine, select,Relationship
"""
設(shè)置數(shù)據(jù)庫參數(shù)
"""
MYSQL_DB = 'xhaccount' #數(shù)據(jù)庫命
MYSQL_USER = 'root' #數(shù)據(jù)庫賬號
MYSQL_PASSWD = '123456' #數(shù)據(jù)庫登陸密碼
MYSQL_HOST = '127.0.0.1' #數(shù)據(jù)庫地址
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) # 數(shù)據(jù)庫連接url
SQLALCHEMY_MIGRATE_REPO = os.path.join(basedir, 'db_repository') # 設(shè)置數(shù)據(jù)庫遷移保存的文件夾，用來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") #關(guān)聯(lián)對象
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") #關(guān)聯(lián)對象
#SQLite不能設(shè)置連接池參數(shù)
engine = create_engine(SQLALCHEMY_DATABASE_URI, echo=True)
def create_db_and_tables():
#創(chuàng)建表
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的數(shù)據(jù)，返回為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()) #顯示第一條數(shù)據(jù) 或者使用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():
#修改表數(shù)據(jù)(通過其他表數(shù)據(jù))
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)
#一次性創(chuàng)建一行
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
)
#分別設(shè)置各行
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)
#獲取查詢結(jié)果(一條)，可以用first()來替代one()
hero_spider_boy = result.one()
statement = select(Team).where(Team.id == hero_spider_boy.id)
result = session.exec(statement)
#獲取查詢結(jié)果記錄集里的第一條
team = result.first()
print("Spider-Boy's team:", team)
print("Spider-Boy's team again:", hero_spider_boy.team)