zoukankan      html  css  js  c++  java

  • Python3之sqlalchemy

    Python3之sqlalchemy

     

         SqlAlchemy是Python编程语言下的一款ORM框架，该框架建立在数据库API之上，使用关系对象映射进行数据库操作，简而言之：将对象转换成SQL，然后使用数据API执行SQL并获取执行结果。

    ORM（对象关系映射）方法论基于三个核心原则：
    • 简单：以最基本的形式建模数据
    • 传达性：数据库结构被任何人都能理解的语言文档化
    • 精准性：基于数据模型创建正确标准化了的结构
     
    Dialect用于和数据API进行交流，根据配置文件的不同调用的数据库API，从而实现对数据库的操作，如：
    '数据库类型+数据库驱动名称://用户名:口令@机器地址:端口号/数据库名'
    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20
    21
    22
    23
    24
    25
    26
    MySQL-Python :
    mysql+mysqldb://<user>:<password>@<host>[:<port>]/<dbname>
     
    pymysql :
    mysql+pymysql://<username>:<password>@<host>/<dbname>[?<options>]
     
    MySQL-Connector :
    mysql+mysqlconnector://<user>:<password>@<host>[:<port>]/<dbname>
     
    cx_Oracle :
    oracle+cx_oracle://user:pass@host:port/dbname[?key=value&key=value...]
     
    更多内容：http://docs.sqlalchemy.org/en/latest/dialects/index.html
    备注：
     
        * Python2.7版本使用mysqldb
        * Python3.5版本使用pymysql
        * 使用pip或者源码包安装，确保环境可以正常使用。
        * 确保远程数据库服务器可以正常使用，并且拥有远程登陆权限
     
    数据库授权：
     
        * mysql -uroot -p           # 登陆数据库
        * create database wang      # 创建数据库
        * grant all on wang.* to wang@"%" identified by '123' # 授权数据库
        * flush privileges          # 更新

    　　

    基本操作

    （1）连接数据库：create_engine()
     
    engine = create_engine("mysql+pymysql://root:123456@127.0.0.1:3306/wang", max_overflow=5)
    • create_engine()   会返回一个数据库引擎
    • mysql+pymysql  “指定一个使用pymysql来连接”
    • 使用用户名'root'和密码'123456'来连接数据库'wang'
    • max_overflow 是最大连接数
    • charset : 设定连接时使用的字符集 charset = utf8
    • echo 参数如果为True时，会显示每条执行的SQL语句，生产环境下课关闭
    （2）字段和数据类型及操作方法
     
    在sqlalchemy.scherma包里有数据库关系的描述，列举几个常用的：
    • 字段 ： Column
    • 索引 ：Index
    • 表：Table
    数据类型在sqlalchemy.types包，列举常用的：
    • 二进制：BIGINT
    • 布尔：BOOLEAN
    • 字符：CHAR
    • 可变字符：VARCHAR
    • 日期：DATATIME
    （3）创建表结构
         使用Schema Type/SQL Expression Language/Engine/ConnectionPooling/Dialect进行数据库操作。Engine使用Schema Type创建一个特定的结构对象，之后通过SQL Expression Language将该对象转换成SQL语句，然后通过ConnectionPooling连接数据库，再然后通过Dialect执行SQL，并获取结果。
     
    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20
    #!/usr/bin/env python3
    #coding:utf8
    from sqlalchemy import create_engine, Table, Column, Integer, String, MetaData, ForeignKey
     
    # 创建数据库连接
    engine = create_engine("mysql+pymysql://root:123456@127.0.0.1:3306/wang", max_overflow=5)
    # 获取元数据
    metadata = MetaData()
    # 定义表
    user = Table('user', metadata,
        Column('id', Integer, primary_key=True),
        Column('name', String(20)),
        )
     
    color = Table('color', metadata,
        Column('id', Integer, primary_key=True),
        Column('name', String(20)),
        )
    # 创建数据表，如果数据表存在，则忽视
    metadata.create_all(engine)

    　　

    继承式增删改查
         使用ORM/Schema Type/SQL Expression Language/Engine/ConnectionPooling/Dialect所有组件对数据库进行操作。根据类创建对象，对象转换成SQL，再执行SQL。
         Query对象可以返回迭代的值（iterator value），然后我们可以通过for in 来查询。不过Query对象的all(), one()以及first()方法将返回非迭代值（non-iterator value），比如说all() 返回的是一个列表, first()方法限制并且仅作为标量返回结果集的第一条记录。
    （1）创建数据库
    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20
    21
    22
    23
    24
    #!/usr/bin/env python
    # -*- coding:utf-8 -*-
    from sqlalchemy.ext.declarative import declarative_base
    from sqlalchemy import Column, Integer, String
    from sqlalchemy.orm import sessionmaker
    from sqlalchemy import create_engine
    # 创建数据库
    engine = create_engine("mysql+pymysql://root:123456@127.0.0.1:3306/wang", max_overflow=5)
    # 生成一个SqlORM 基类
    Base = declarative_base()
    # 定义表结构
    class User(Base):
        # 表名
        __tablename__ = 'users'
        # 定义id,主键唯一,
        id = Column(Integer, primary_key=True)
        name = Column(String(50))
    # 寻找Base的所有子类，按照子类的结构在数据库中生成对应的数据表信息
    Base.metadata.create_all(engine)
    # 创建与数据库的会话session class ,注意,这里返回给session的是个class,不是实例
    Session = sessionmaker(bind=engine)
    session = Session()
    # 获取session，然后把对象添加到session
    # 最后提交并关闭。Session对象可视为当前数据库连接。

    　　（2）增加

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20
    21
    22
    23
    24
    25
    26
    27
    28
    29
    30
    #!/usr/bin/env python
    # -*- coding:utf-8 -*-
    from sqlalchemy.ext.declarative import declarative_base
    from sqlalchemy import Column, Integer, String
    from sqlalchemy.orm import sessionmaker
    from sqlalchemy import create_engine
     
    engine = create_engine("mysql+pymysql://root:123456@127.0.0.1:3306/wang", max_overflow=5)
    Base = declarative_base()
     
    class User(Base):
        __tablename__ = 'users'
        id = Column(Integer, primary_key=True)
        name = Column(String(50))
     
    Base.metadata.create_all(engine)
    Session = sessionmaker(bind=engine)
    session = Session()
    ####  添加 ########
    # 定义一个字段
    zengjia = User(id=2, name='sbliuyao')
    # 添加字段
    session.add(zengjia)
    # 添加多个字段
    session.add_all([
        User(id=3, name='sbyao'),
        User(id=4, name='liuyao')
    ])
    # 提交以上操作，现在只是在内存中增加，回写到数据库，就必须做提交操作
    session.commit()

    　　（3）删除

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20
    21
    22
    23
    #!/usr/bin/env python
    # -*- coding:utf-8 -*-
    from sqlalchemy.ext.declarative import declarative_base
    from sqlalchemy import Column, Integer, String
    from sqlalchemy.orm import sessionmaker
    from sqlalchemy import create_engine
     
    engine = create_engine("mysql+pymysql://root:123456@127.0.0.1:3306/wang", max_overflow=5)
    Base = declarative_base()
     
    class User(Base):
        __tablename__ = 'users'
        id = Column(Integer, primary_key=True)
        name = Column(String(50))
     
    Base.metadata.create_all(engine)
    Session = sessionmaker(bind=engine)
    session = Session()
     
    ########### 删除 ##########
    # 删除user表，id大于2的字段
    session.query(User).filter(User.id > 2).delete()
    session.commit()

    　　（4）修改

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20
    21
    #!/usr/bin/env python
    # -*- coding:utf-8 -*-
    from sqlalchemy.ext.declarative import declarative_base
    from sqlalchemy import Column, Integer, String
    from sqlalchemy.orm import sessionmaker
    from sqlalchemy import create_engine
     
    engine = create_engine("mysql+pymysql://root:123456@127.0.0.1:3306/wang", max_overflow=5)
    Base = declarative_base()
     
    class User(Base):
        __tablename__ = 'users'
        id = Column(Integer, primary_key=True)
        name = Column(String(50))
     
    Base.metadata.create_all(engine)
    Session = sessionmaker(bind=engine)
    session = Session()
    # user表里的id等于2的字段修改为id=6
    session.query(User).filter(User.id == 2).update({'id' : 6})
    session.commit()

    　　（5）查询

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20
    21
    22
    23
    24
    25
    26
    27
    28
    29
    30
    31
    32
    33
    34
    35
    36
    37
    38
    39
    40
    41
    42
    43
    44
    45
    46
    47
    ###　查询方式1 ####
    # 查询User表中字段是name=liuyao的第一条数据
    ret = session.query(User).filter_by(name='liuyao').all()
    print(ret)  # 输出ret，这是个对象的内存地址
    for i in ret:
        print(i.id,i.name)  # 输出ret的内容
     
    ###　查询方式2 ####
    # 查询user表里字段是name=liuyao的第一条数据
    ret = session.query(User).filter_by(name='liuyao').first()
    print(ret)        # 输出的结果为对象的内存地址
    print(ret.name)   # 输出结果的name字段
    print(ret.id)     # 输出结果的id字段
     
    ### 查询方式3 ###
    # 查询user表里字段是name是liuyao或者mayun的信息打印出来
    ret = session.query(User).filter(User.name.in_(['liuyao','mayun'])).all()
    print(ret)
    for i in ret:
        print(i.name,i.id)
     
    ### 查询方式4 ###
    # 可以给返回的结果起一个别名，或者叫标签：可有可无
    ret = session.query(User.name.label('')).all()
    # 这里的关键是label方法，它的意思是把User的name字段改个名字叫name_label,
    # 其相当于执行了：select users.name as name_label from User
    print(ret,type(ret))
     
    ### 查询方式5 ###
    # 查询User表根据id排序
    ret = session.query(User).order_by(User.id).all()
    print(ret)
    for i in ret:
        print(i.name)
     
    ### 查询方式6 ###
    # 查询user表里根据id排序输入0到3的字段
    ret = session.query(User).order_by(User.id)[0:3]
    print(ret)
    for i in ret:
        print(i.name)
     
    ### 查询方式7 ###
    # 创建Query查询，filter是where条件，最后调用one()返回唯一行，如果调用all()则返回所有行
    user = session.query(User).filter(User.id=='5').one()
    print(type(user))  # 查看user的类型
    print(user.name)   # 查看对象的name属性

    　　（6）外键关联

     由于关系型数据的多个表还可以用外键实现一对多，多对多等关联，相应的，ORM框架也可以提供两个对象之间的一对多，多对多等功能，
    • 一对多（一个User可以有多个Address）
    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20
    21
    22
    23
    24
    25
    26
    27
    28
    29
    30
    31
    32
    33
    34
    35
    36
    37
    38
    39
    40
    41
    42
    43
    44
    45
    46
    47
    48
    49
    50
    51
    52
    53
    54
    #!/usr/bin/env python3
    #coding:utf8
    # 导入所需模块
    from sqlalchemy import create_engine,func
    from sqlalchemy.ext.declarative import declarative_base
    from sqlalchemy import Column, Integer, String,ForeignKey
    from  sqlalchemy.orm import sessionmaker,relationship
    # 生成sqlorm基类
    Base = declarative_base()
    # 创建数据库连接
    engine = create_engine("mysql+pymysql://root:123456@127.0.0.1:3306/wang", max_overflow=5)
    # 目的是一个人可以拥有多本书，那么在数据库里的一对多关系
    class User(Base):
        # 表名
        __tablename__ = 'user'
        # id字段
        id = Column(String(20), primary_key=True)
        # 名字字段
        name = Column(String(20))
        # 一对多:
        # 内容不是表名而是定义的表结构名字
        books = relationship('Book')
    class Book(Base):
        # 表明
        __tablename__ = 'book'
        # id字段
        id = Column(String(20), primary_key=True)
        # 名字字段
        name = Column(String(20))
        # “多”的一方的book表是通过外键关联到user表的:
        # ForeignKey是外键 关联user表的id字段
        user_id = Column(String(20), ForeignKey('user.id'))
    # 创建所需表
    Base.metadata.create_all(engine)
     
    if __name__ == '__main__':
        # 绑定,生成会话
        SessionCls = sessionmaker(bind=engine)
        session = SessionCls()
        # 创建用户
        liuyao = User(id='1',name='liuyao')
        ali = User(id='2',name='ali')
        # 添加字段
        session.add_all([liuyao,ali])
        # 提交
        session.commit()
        # 创建白鹿原这本书，指定谁是拥有者
        Whitedeer = Book(id='1',name='White_deer',user_id = '1')
        # 创建三体这本书，指定谁是拥有者
        Threebody = Book(id='2',name='Three_body',user_id = '2')
        # 添加字段
        session.add_all([Whitedeer,Threebody])
        # 提交
        session.commit()

    　　

    • 多对多
    建立一个双向一对多关系，“反向”是一个许多人，指定一个额外的relationship()函数，并连接两个使用relationship.back_populates参数。简单来说，relationship函数是sqlalchemy对关系之间提供的一种便利的调用方式，backref参数则对关系提供反向引用的声明。在最新版的sqlalchemy中对relationship引进了back_populates参数。
    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20
    21
    22
    23
    24
    25
    26
    27
    28
    29
    30
    31
    32
    33
    34
    #!/usr/bin/env python3
    #coding:utf8
    from sqlalchemy import Column, Sequence, String, Integer, ForeignKey
    from sqlalchemy import create_engine # 导入创建连接驱动
    from sqlalchemy.ext.declarative import declarative_base
    from sqlalchemy.orm import sessionmaker
    from sqlalchemy.orm import relationship, backref
    # 这个url可以用urlparse解析, 其中echo=True表示执行时显示sql语句
    engine = create_engine("mysql+pymysql://root:123456@127.0.0.1:3306/wang", max_overflow=5)
    # 生成了declarative基类, 以后的model继承此类
    Base = declarative_base()
    class Parent(Base):
        __tablename__ = 'parent'
        id = Column(Integer, primary_key=True)
        name = Column(String(64),unique=True,nullable=False)
        children = relationship("Child", back_populates="parent")
    class Child(Base):
        __tablename__ = 'child'
        id = Column(Integer, primary_key=True)
        name = Column(String(64),unique=True,nullable=False)
        parent_id = Column(Integer, ForeignKey('parent.id'))
        parent = relationship("Parent", back_populates="children")
    Base.metadata.create_all(engine) # 创建所有表结构
    if __name__ == '__main__':
        SessionCls = sessionmaker(bind=engine)
        # 创建与数据库的会话session class ,注意,这里返回给session的是个class,不是实例
        session = SessionCls()
        mama = Parent(id='1',name='mamaxx')
        baba = Parent(id='2',name='babaoo')
        session.add_all([mama,baba])
        onesb = Child(id='1',name='onesb',parent_id='2')
        twosb = Child(id='2',name='twosb',parent_id='2')
        session.add_all([onesb,twosb])
        session.commit()
    • 多对多之三表外键关联
    • 1
      2
      3
      4
      5
      6
      7
      8
      9
      10
      11
      12
      13
      14
      15
      16
      17
      18
      19
      20
      21
      22
      23
      24
      25
      26
      27
      28
      29
      30
      31
      32
      33
      34
      35
      36
      37
      38
      #!/usr/bin/env python3
      #coding:utf8
      from sqlalchemy import create_engine,func,Table
      from sqlalchemy.ext.declarative import declarative_base
      from sqlalchemy import Column, Integer, String,ForeignKey
      from sqlalchemy.orm import sessionmaker,relationship
      Base = declarative_base()
      # 关系表
      Host2Group = Table('host_2_group',Base.metadata,
                    Column('host_id',ForeignKey('hosts.id'),primary_key=True),
                    Column('group_id',ForeignKey('group.id'),primary_key=True),
                    )
      engine = create_engine("mysql+pymysql://root:123456@127.0.0.1:3306/wang", max_overflow=5)
      class Host(Base):
          __tablename__ = 'hosts'
          id = Column(Integer,primary_key=True,autoincrement=True)
          hostname = Column(String(64),unique=True,nullable=False)
          ip_addr = Column(String(128),unique=True,nullable=False)
          port = Column(Integer,default=22)
          groups = relationship('Group',
                            secondary= Host2Group,
                            backref = 'host_list')
      class Group(Base):
          __tablename__ = 'group'
          id = Column(Integer,primary_key=True)
          name = Column(String(64),unique=True,nullable=False)
       
      Base.metadata.create_all(engine) # 创建所有表结构
       
      if __name__ == '__main__':
          SessionCls = sessionmaker(bind=engine)
          session = SessionCls()
          g1 = Group(name='g1')
          g2 = Group(name='g2')
          g3 = Group(name='g3')
          g4 = Group(name='g4')
          session.add_all([g1,g2,g3,g4])
          session.commit()

      　
  • 相关阅读:
    [学习记录]Prometheus简单上手
    [学习记录]从triggerflow源码中学到的简单思想
    42 | 实战：大型全球化电商的测试基础架构设计
    41 | 从小工到专家：聊聊测试执行环境的架构设计（下）
    40 | 从小工到专家：聊聊测试执行环境的架构设计（上）
    39 | 从小作坊到工厂：什么是Selenium Grid？如何搭建Selenium Grid？
    python之jsonpath
    38 | 测试数据的“银弹”- 统一测试数据平台（下）
    Python replace()方法
    37 | 测试数据的“银弹”- 统一测试数据平台（上）
  • 原文地址：https://www.cnblogs.com/meng-wei-zhi/p/8259019.html
Copyright © 2011-2022 走看看