SQLAlchemy基本使用

清疚 2023-08-17 16:53 108阅读 0赞

## 简单介绍SQLAlchemy ##

> **sqlAlchemy是python中最著名的ORM(Object Relationship Mapping)框架了。**
> 
> **什么是ORM？**
> 
> **一句话解释的话就是，一种可以把model中的模型和数据库中的一条数据相互转换的工具。**

## **初始化数据库连接** ##

# 初始化数据库链接
    engine = create_engine("mysql+pymysql://root:123@127.0.0.1:3306/sqlalchemy?charset=utf8")

> create\_engine()函数中的字符串的意义是，数据库+数据库连接框架://用户名:密码@IP地址:端口号/数据库名称。  
> 那么不难看出，我使用了mysql数据库，数据库连接用的是pymysql，用户名root，密码是123，ip地址是127.0.0.1，端口号3306(这是mysql服务器默认端口号)，sqlalchemy是数据库的名字,charset是设置编码。

首先需要准备安装：

pip install sqlalchem
    pip install pymysql

## 创建Users模型 ##

from sqlalchemy.ext.declarative import declarative_base
    
    # 建立基础类 R关系 M映射 类
    Model = declarative_base()
    
    from sqlalchemy import Column
    from sqlalchemy import Integer
    from sqlalchemy import String
    
    
    # 定义User对象
    class Users(Model):
        # 表的名字
        __tablename__ = "user"
        # 表的结构
        id = Column(Integer, primary_key=True, autoincrement=True)
        # nullable 是否允许为空
        name = Column(String(32), nullable=False)
    
    
    from sqlalchemy.engine import create_engine
    
    engine = create_engine("mysql+pymysql://root@127.0.0.1:3306/sqlalchemy?charset=utf8")
    
    # 检索所有继承 Model 的Object 并在 engine 指向的数据库中创建 所有的表
    Model.metadata.create_all(engine)
    # Model.metadata.drop_all(engine)

## 单表的CRUD ##

> ** CURD（Create Read Update Delete，增删改查）**
> 
> **在sqlAlchemy中，增删改查操作是通过一个session对象(DBSession,是由sessionmaker创建的)来完成的。**

from sqlalchemy.orm import sessionmaker
    from s1 import engine, Users  # 导入之前创建的engine和Users
    
    session = sessionmaker(engine)  # 新建数据库的查询窗口
    db_session = session()  # 打开查询窗口

**增删改查之前都需要有查询窗口**

** 增**

================增加单条数据===============
    user = Users(name="周杰伦")  # 新建insert语句 insert into
    db_session.add(user)  # 将insert语句移动到 db_session 查询窗口
    db_session.commit()  # 执行查询窗口中的所有语句
    db_session.close()  # 关闭查询窗口

=================增加多条数据===================
    user_list = [Users(name="小红"), Users(name="小明")]
    db_session.add_all(user_list)  # 将所有的insert 语句移动到 查询窗口
    db_session.commit()
    db_session.close()

![1614698-20190731155808957-973737269.png][]

**查**

# 查询数据
    res = db_session.query(Users).all()  # 查询所有数据
    for user in res:
        print(user.id, user.name)
    
    # 打印内容
    1 周杰伦
    2 小红
    3 小明

res = db_session.query(Users).first()  # 查询符合条件的第一条数据
    print(res.id, res.name)
    
    # 打印内容
    1 周杰伦

# 简单带条件的查询
    res = db_session.query(Users).filter(Users.id < 3).all()
    print(res)
    for user in res:
        print(user.id, user.name)
    
    # 打印内容
    [<s1.Users object at 0x00000204DF431320>, <s1.Users object at 0x00000204DF431390>]
    1 周杰伦
    2 小红

# 并列条件
    res = db_session.query(Users).filter(Users.id < 3, Users.name == "周杰伦").all()
    for user in res:
        print(user.id, user.name)
    
    res = db_session.query(Users).filter(Users.id < 3, Users.name == "周杰伦").first()
    print(res.id, res.name)
    
    # 打印内容都是
    1 周杰伦

**改（更新**）

# 修改更新数据
    db_session.query(Users).filter(Users.id == 2).update({
         "name": "小黑"})
    # 更新的sql语句
    db_session.commit()

**删**

# 删除数据
    db_session.query(Users).filter(Users.id == 3).delete()
    db_session.commit()

##  一对多的CRUD ##

** 首先创建Student和School**

# 和单表类似
    from sqlalchemy.ext.declarative import declarative_base
    
    # 基类
    Model = declarative_base()
    
    from sqlalchemy import Column, Integer, String, ForeignKey
    from sqlalchemy.orm import relationship
    
    
    class Student(Model):
        __tablename__ = "student"
        id = Column(Integer, primary_key=True)
        name = Column(String(32), nullable=False)
        sch_id = Column(Integer, ForeignKey("school.id"))  # 多对一关系存储列
        # relationship ORM精髓所在
        stu2sch = relationship("School", backref="sch2stu")
    
    
    class School(Model):
        __tablename__ = "school"
        id = Column(Integer, primary_key=True)
        name = Column(String(32), nullable=False)
    
    
    from sqlalchemy.engine import create_engine
    
    engine = create_engine("mysql+pymysql://root@127.0.0.1:3306/sqlalchemy?charset=utf8")
    
    Model.metadata.create_all(engine)

**增删改查之前也需要创建窗口的**

# 同样的流程
    from sqlalchemy.orm import sessionmaker
    from 多对一查询 import engine, School, Student
    
    session = sessionmaker(engine)
    db_session = session()

**增**

# 增加一条数据 # 优先增加school
    sc = School(name="清华大学") db_session.add(sc) db_session.commit() # 再添加 student
    sch_first = db_session.query(School).filter(School.name == "清华大学").first() stu = Student(name="周杰伦", sch_id=sch_first.id) db_session.add(stu) db_session.commit()

以上方法也是可以增加数据成功的，但是有更简单的方法。

# 添加数据 之 relationship 正向添加数据
    stu = Student(name="小红", stu2sch=School(name="清华大学")) db_session.add(stu) db_session.commit() # 添加数据 之 relationship 反向添加数据
    sch = School(name="北京大学") sch.sch2stu = [ Student(name="小吴"), Student(name="小侯") ] db_session.add(sch) db_session.commit()

**查**

# 查询 的 relationship 正向
    res = db_session.query(Student).all() for stu in res: print(stu.id, stu.name, stu.stu2sch.name) # 打印内容
    1 周杰伦 清华大学 2 小红 清华大学 3 小吴 北京大学 4 小侯 北京大学

# 查询 的 relationship 反向
    res = db_session.query(School).all() for sch in res: # print(sch.name,len(sch.sch2stu))
        for stu in sch.sch2stu: print(sch.name, stu.name) # 打印内容
    清华大学 周杰伦 清华大学 小红 北京大学 小吴 北京大学 小侯

##  多对多的CURD ##

** 创建多对多关系，比如书籍和作者**

from sqlalchemy.ext.declarative import declarative_base
    
    Model = declarative_base()
    
    from sqlalchemy import Column, Integer, String, ForeignKey
    from sqlalchemy.orm import relationship
    
    
    class Books(Model):
        __tablename__ = "book"
        id = Column(Integer, primary_key=True)
        name = Column(String(32), nullable=False)
        # relationship  backref是反向关联的关键字 secondary 指定多对多中记录的排序方式
        b2a = relationship("Authors", backref="a2b", secondary="book_author")
    
    
    class Authors(Model):
        __tablename__ = "author"
        id = Column(Integer, primary_key=True)
        name = Column(String(32), nullable=False)
    
    
    class Book_author(Model):
        __tablename__ = "book_author"
        id = Column(Integer, primary_key=True)
        book_id = Column(Integer, ForeignKey("book.id"))
        author_id = Column(Integer, ForeignKey("author.id"))
    
    
    from sqlalchemy.engine import create_engine
    
    engine = create_engine("mysql+pymysql://root@127.0.0.1:3306/sqlalchemy?charset=utf8")
    
    Model.metadata.create_all(engine)
    # Model.metadata.drop_all(engine)

**增**

from sqlalchemy.orm import sessionmaker
    from 多对多 import engine, Books, Authors
    
    s = sessionmaker(engine)
    db_session = s()
    
    # 添加数据 之 relationship 正向添加数据
    book = Books(name="西游记", b2a=[Authors(name="吴承恩"), Authors(name="周杰伦")])
    db_session.add(book)
    db_session.commit()
    
    # 添加数据 之 relationship 反向添加数据
    a = Authors(name="周杰伦")
    a.a2b = [
        Books(name="以父之名"),
        Books(name="退后")
    ]
    db_session.add(a)
    db_session.commit()

**查**

# 查询数据 之 relationship 正向
    res = db_session.query(Books).all()
    for book in res:
        for author in book.b2a:
            print(book.name, author.name)
    
    # 打印内容
    西游记 吴承恩
    西游记 周杰伦
    以父之名 周杰伦
    退后 周杰伦
    
    # 查询数据 之 relationship 反向
    res = db_session.query(Authors).all()
    for author in res:
        for book in author.a2b:
            print(book.name,author.name)
    
    # 打印内容
    西游记 吴承恩
    西游记 周杰伦
    以父之名 周杰伦
    退后 周杰伦

## 高级版查询操作 ##

#### and和or操作 ####

from s1 import engine, Users as User
    from sqlalchemy.orm import sessionmaker
    
    Session = sessionmaker(engine)
    db_session = Session()
    
    from sqlalchemy import and_, or_
    
    ret = db_session.query(User).filter(or_(User.id == 2, User.name == '周杰伦')).all()
    
    print([i.name for i in ret])
    
    # 打印内容
    ['周杰伦', '小黑']

#### or 条件中 有两个 并且and ####

ret = db_session.query(User).filter(
        or_(
            and_(User.id == 1, User.name == '周杰伦'),
            and_(User.id == 2, User.name == '小黑')
        )
    ).all()
    
    print(ret)
    
    # 打印内容
    [<s1.Users object at 0x000001BAC9DC09E8>, <s1.Users object at 0x000001BAC9DC0B00>]

.all()去掉就是SQL语句

#### order\_by排序 ####

# 默认是升序
    user_list = db_session.query(User).order_by(User.id).all()
    user_list = db_session.query(User).order_by(User.id.desc()).all()
    for row in user_list:
        print(row.name,row.id)
    
    # 打印内容
    小黑 2
    周杰伦 1

#### 查询数据 指定查询数据列 加入别名 ####

r2 = db_session.query(User.name.label('username'), User.id).first()
    print(r2.id, r2.username)  # 1 周杰伦

#### 表达式筛选条件 ####

r3 = db_session.query(User).filter(User.name == "周杰伦").all()
    print(r3)
    
    # [<s1.Users object at 0x000001CE0D1AFC88>]

#### 原生SQL筛选条件 where name="周杰伦" ####

r4 = db_session.query(User).filter_by(name='周杰伦').all()
    r5 = db_session.query(User).filter_by(name='周杰伦').first()

#### 原生SQL查询（不常用，还不如使用pymysql） ####

from sqlalchemy import text
    r7 = db_session.query(User).from_statement(text("SELECT * FROM User where name=:name")).params(name='周杰伦').all()
    print(r7)
    
    # [<s1.Users object at 0x00000214C1EFF0F0>]

#### 筛选查询列 ####

# query的时候我们不在使用User ORM对象,而是使用User.name来对内容进行选取
    user_list = db_session.query(User.name).all()
    print(user_list)
    for row in user_list:
        print(row.name)
    
    #打印内容
     [('周杰伦',), ('小黑',)]
    周杰伦
    小黑

#### 别名映射 name as nick ####

user_list = db_session.query(User.name.label("nick")).all()
    print(user_list)
    for row in user_list:
        print(row.nick) # 这里要写别名了
    
    # 打印内容
    [('周杰伦',), ('小黑',)]
    周杰伦
    小黑

####  复杂查询 ####

from sqlalchemy.sql import text
    user_list = db_session.query(User).filter(text("id<:value and name=:name")).params(value=3,name="周杰伦").all()
    print(user_list)

#### 其他查询条件 ####

ret = db_session.query(User).filter(User.id.between(1, 3), User.name == '周杰伦').all()  # between 大于1小于3的
    print(ret)
    
    # 打印内容
    [<s1.Users object at 0x0000019F9EA10F28>]

r1 = db_session.query(User).filter(~User.id.between(1,3)).all()
    print(r1)
    
    # 打印内容
    []

ret = db_session.query(User).filter(User.id.in_([1,3,4])).all() # in_([1,3,4]) 只查询id等于1,3,4的
    print(ret)

r1 = db_session.query(User).filter(~User.id.in_([1,3,4])).all()
    print(r1[0].id)
    
    # 打印内容
    2

ret = db_session.query(User).filter(~User.id.in_([1, 3, 4])).all()  # ~xxxx.in_([1,3,4]) 查询不等于1,3,4的
    print(ret)

ret = db_session.query(User).filter(User.id.in_(db_session.query(User.id).filter_by(name='周杰伦'))).all() # 子查询
    print(ret)
    
    # 打印内容
    [<s1.Users object at 0x00000196789EFE10>]

#### 通配符 ####

ret = db_session.query(User).filter(User.name.like('小%')).all()
    ret = db_session.query(User).filter(~User.name.like('小%')).all()

#### 限制 ####

ret = db_session.query(User)[1:2]
    res = db_session.query(User)[:2]

#### 分组 ####

ret = db_session.query(User).group_by(User.name).all()

from sqlalchemy.sql import func
    
    ret = db_session.query(
        func.max(User.id),
        func.sum(User.id),
        func.min(User.id)).group_by(User.name).all()
    print(ret)
    
    # 打印内容
    [(1, Decimal('1'), 1), (2, Decimal('2'), 2)]

from sqlalchemy.sql import func
    
    ret = db_session.query(
        func.max(User.id),
        func.sum(User.id),
        func.min(User.id)).group_by(User.name).having(func.min(User.id) >2).all()
    print(ret)
    
    # 打印内容
    []

转载于:https://www.cnblogs.com/biao-wu/articles/11276672.html

[1614698-20190731155808957-973737269.png]: /images/20230809/4febdaf33f58424fb1e7aec54da37989.png