Benjamin

静以修身,俭以养德,非澹薄无以明志,非宁静无以致远。
随笔 - 386, 文章 - 0, 评论 - 196, 引用 - 0
数据加载中……

SQLModel对mysql基本操作

SQLModel,Python中的SQL数据库,旨在实现简单性、兼容性和健壮性。
Documentation: https://sqlmodel.tiangolo.com
Source Code: https://github.com/tiangolo/sqlmodel
兼容 FastAPI, Pydantic, SQLAlchemy.可以使用SQLAlchemy、Pydantic里的大部分接口
安装时自动安装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)

posted on 2021-09-04 15:08 Benjamin 阅读(1690) 评论(0)  编辑 收藏 引用 所属分类: python


只有注册用户登录后才能发表评论。
网站导航: 博客园   IT新闻   BlogJava   知识库   博问   管理