[大数据]【python】sqlalchemy

SQLAlchemy

安装

pip install SQLAlchemy

查看版本号

import sqlalchemy
print(sqlalchemy.__version__)

建立连接Engine

from sqlalchemy import create_engine
engine = create_engine(
    'mysql+pymysql://127.0.0.1:3306/wss_demmo?charset=utf8&user=root&password=123456',
    echo=True,  # 打印日志
    future=True  # 新版本接口格式
)

执行SQL

查询数据

from sqlalchemy import create_engine,text
with engine.connect() as conn:
    result = conn.execute(text("select * from test"))
    print(result.all())

逐行读取数据

with engine.begin() as conn:
    result = conn.execute(text("SELECT * FROM TEST"));
    for row in result:
        print(f"id:{row.id},name:{row.name}")

默认返回的是元组类型，可以通过索引和属性名称来获取每一组的数据

使用??result.mappings（）?修饰符??转换为??MapingResult??对象，这样一来集合的成员类型都是字典类型。

with engine.begin() as conn:
    result = conn.execute(text("SELECT * FROM TEST"));
    for row in result.mappings():
        print(row)  # {'id': '8', 'name': '赵六'}

传递参数

with engine.begin() as conn:
    result = conn.execute(text("SELECT * FROM TEST WHERE id > :id"),{"id":5});
    print(result.all())

参数使用:参数名来指定，并在后面的字典参数中来进行传值。如 :id与后面的字典就是对应关系。

当需要传递多组参数，执行多次语句时，可以传递字典列表来实现，如下面的保存示例。

提交更改

默认事务在执行以后会自动回滚，如果需要提交数据，需要调用conn.commit();

with engine.connect() as conn:
    result = conn.execute(text("INSERT INTO test (id, name) VALUES (:id, :name)"),
                          [{"id": 5, "name": "赵六"}, {"id": 6, "name": "张七"}]);
    conn.commit()

还有另一种提交方法，使用begin来获取连接，这种会在事务结束时自动提交

with engine.begin() as conn:
    result = conn.execute(text("INSERT INTO test (id, name) VALUES (:id, :name)"),
                          [{"id": 8, "name": "赵六"}, {"id": 7, "name": "张七"}]);

ORM

ORM会话执行SQL

from sqlalchemy.orm import Session
stmt = text("SELECT * FROM TEST WHERE id > :id").bindparams(id=3)
with Session(engine) as session:
    result = session.execute(stmt);
    print(result.all())

同connection一样，进行插入和修改操作时需要提交（session.commit()）才可以生效。

with Session(engine) as session:
    result = session.execute(text("UPDATE TEST SET  name=:name where id=:id"), [{"name": "展招", "id": 5}, {"name": "包大人", "id": 6}]);
    session.commit()

元数据MetaData

创建元数据对象

from sqlalchemy import MetaData
metadata_obj = MetaData()

Table对象

from sqlalchemy import Table, Column, Integer, String
user_table = Table(
    "user_account",
    metadata_obj,
    Column('id', Integer, primary_key=True),  # 主键约束
    Column('name', String(30)),
    Column('fullname', String(50))
)

简单约束

from sqlalchemy import ForeignKey
address_table = Table(
    "address",
    metadata_obj,
    Column('id', Integer, primary_key=True),  # 主键约束
    Column('user_id', ForeignKey('user_account.id'), nullable=False),  # ForeignKey：外键约束   nullable：非空约束
    Column('email_address', String(100), nullable=False)
)

在??列?定义中使用??ForeignKey?对象时，我们可以省略该??列??的数据类型 ;它是从相关列的自动推断出来的，在上面的示例中，列的??Integer?数据类型。?user_account.id

查看表对象的主键约束

print(user_table.primary_key)

DDL

元数据对象通过已经关联的table对象生成数据库表。

metadata_obj.create_all(engine)

MetaData??对象还具有一个??MetaData.drop_all（）??方法，该方法将删除所有元数据对象关联的数据库表。

新建ORM模型

通过注册获取元数据基类

from sqlalchemy.orm import registry

mapper_registry = registry()
Base = mapper_registry.generate_base()

还提供了一个方法可以直接获取到基类

from sqlalchemy.orm import declarative_base
Base = declarative_base()

创建表

# 通过注册获取的元数据来创建表
mapper_registry.metadata.create_all(engine)
# 通过基类的元数据创建表
Base.metadata.create_all(engine)

现在就可以根据基类来映射类了

from datetime import datetime
import pymysql
from sqlalchemy import create_engine
from sqlalchemy.orm import declarative_base
from sqlalchemy import Column, Integer, String, DateTime, Enum, SmallInteger, Boolean
from enum import IntEnum
from sqlalchemy.dialects.mysql import TINYINT, CHAR

pymysql.install_as_MySQLdb()  # 解决3.*版本mysqlDb模块找不到问题
engine = create_engine(
    'mysql://127.0.0.1:3306/wss_demmo?charset=utf8&user=root&password=123456',
    echo=True,  # 打印日志
    future=True  # 新版本接口格式
)

Base = declarative_base()

class SexEnum(IntEnum):
    MAN = 1
    WOMAN = 0

class Student(Base):
    __tablename__ = "student"
    id = Column(type_=Integer, primary_key=True, name="id", autoincrement=True)
    stu_no = Column(type_=Integer, nullable=False, comment="学号", doc="只在代码中能看到的注释", unique=True)
    stu_name = Column(CHAR(16), nullable=False, comment="姓名", default="默认值")
    sex = Column(Enum(SexEnum), default=None, comment="性别")
    age = Column(TINYINT(unsigned=True), default=0, comment="年龄")
    create_at = Column(type_=DateTime, default=datetime.now())
    is_vaild = Column(Boolean, default=True)

if __name__ == '__main__':
    Base.metadata.create_all(bind=engine)  # 根据继承Base的实体类生成表
    # Base.metadata.drop_all(bind=engine)  # 根据继承Base的实体类删除表

关联

一对多关联

from sqlalchemy.orm import declarative_base, relationship
from sqlalchemy import Column, Integer, String, DateTime, Enum, SmallInteger, Boolean, ForeignKey
class Address:
    __tablename__ = "address"
    stu_id = Column(Integer, ForeignKey(Student.id))  # 外键
    user = relationship("User", backref="address")  # 一对多

一对一关联

user = relationship("User", backref="address",uselist=False)  # 一对一

创建Session

from sqlalchemy.orm import Session
session = Session(bind=engine, future=True)

插入

插入单行

from sqlalchemy.orm import Session
session = Session(bind=engine, future=True)
print("连接")
with session:
    stu = Student(id=3, stu_no=3, stu_name="李四", sex=SexEnum.MAN)
    Address(user=stu)
    session.add(stu)
    session.commit()

插入多行

    print("连接")
    with session:
        stu_list = []
        addrs = []
        for i in range(10):
            stu = Student(stu_no=i, stu_name="李四", sex=SexEnum.MAN)
            addrs.append(Address(user=stu))
            stu_list.append(stu)
        session.add_all(addrs)
        session.add_all(stu_list)
        session.commit()

查询

主键查询

session.get(Entity,key);

select查询

stmt = select(Student.id,Student.sex)
rows = session.execute(stmt).fetchall()
for row in rows:
	print(row)

条件查询

stmt = select(Student).where(Student.id == 1)
rows = session.execute(stmt)
print(rows.scalars().one())  # 只有查询结果只有一条时才可以使用
print(rows.scalar_one())