SQLAlchemy的使用

Posted 2021-01-19 Python爱好者666

tags:

篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了SQLAlchemy的使用相关的知识，希望对你有一定的参考价值。

Orm简介：

对象关系映射，对象和关系之间的映射

Table >>> class 表类

Row >>> object 行实例（每一行都是一个个体，就是一个个实例）

Column >>>> property 列属性。

1、sqlalchemy的简介

Sqlalchemy是一个orm框架。大量使用了元编程。

安装pip install SQLalchemy。

import sqlalchemy
print(sqlalchemy.__version__)

版本查看：

Sqlalchemy的官方文档http://docs.sqlalchemy.org/en/latest/

Sqlalchemy内部使用了连接池

2、创建连接

1）创建引擎

⑴mysqldb连接方式：

Mysql+mysqldb：//<user>:<password>@<host>[:<port>/<dbname>]

engine=sqlalchmety.create_engine(“mysql+mysqldb:wang:[email protected]:3306/mysql”)

⑵pymysql

Mysql+pymysql://<user>:<password>@<host>/<dbname>[?<options>]

engine=sqlalchemy.create_engine(“mysql+pymysql:wang:[email protected]:3306/test”)

engine= sqlalchemy.create_engine(“mysql+pymysql:wang:[email protected]:3306/test”,echo=True)

echo=True引擎是否打印执行的语句。

方言+pymysql。

2）创建映射declare a mapping

from sqlalchemy.ext.declarative import declarative_base

创建基类Base = declarative_base()

3）创建实体类

代码中的自增不会控制数据库中的表，所以写不写都可以的，数据是对应数据库的，自增没有任何关系。

import sqlalchemy
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column,Integer,String,Date,Enum

connstr = "{},{},{},{},{}".format(‘msql+pymsql‘,‘wang‘,‘wang‘,‘192.168.118.145‘,‘school‘)
engine = sqlalchemy.create_engine(connstr,echo=True)

Base = declarative_base()

class Student(Base): #创建实体类
    #指定表明
    __tablename__ = ‘student‘
    id = Column(Integer,primary_key=True)
    name = Column(String(64))
    age = Column(Integer) #第一参数是字段名，如果和属性名不一致，要指定 age=Column(‘age‘,Integer)

    def __repr__(self):
        return "{}id={}name={}age={}".format(self.__class__.__name__,self.id,self.name,self.age)

4）实例化

import sqlalchemy
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column,Integer,String,Date,Enum

host = ‘192.168.118.145‘
user = ‘wang‘
password = ‘wang‘
port = 3306
database = ‘school‘

connstr = "mysql+pymysql://{}:{}@{}:{}/{}".format(user,password,host,port,database)
engine = sqlalchemy.create_engine(connstr,echo=True)

Base = declarative_base()

class Student(Base): #创建实体类
    #指定表明
    __tablename__ = ‘student‘
    id = Column(Integer,primary_key=True)
    name = Column(String(64))
    age = Column(Integer) #第一参数是字段名，如果和属性名不一致，要指定 age=Column(‘age‘,Integer)

    def __repr__(self):
        return "{}id={}name={}age={}".format(self.__class__.__name__,self.id,self.name,self.age)

s1 = Student(name=‘tom‘)
print(s1)
print(s1.name)
s1.age=23
print(s1.age)

Column的类型必须是写进去的。

Tablename是指定表名，column中必须写。

CREATE TABLE student (

id INTEGER NOT NULL AUTO_INCREMENT,

name VARCHAR(64),

age INTEGER,

PRIMARY KEY (id)

)

5）创建表（少用）

利用sqlalchemy来创建表、删除表

Base.metadata.drop_all(engine) #删除继承自Base的所有的表
Base.metadata.create_all(engine) #创建所有继承自Base的所有的表

这两种方法少用。

6）建立会话

from sqlalchemy.orm import sessionmaker

#6建立会话
Session = sessionmaker(bind=engine)#返回类
session = Session()

建立会话：Session = sessionmaker(bind=engine) session = Session()

session对象线程不安全，所以不同线程使用不同的session对象，

Session类和engine都是线程安全的，有一个就行了。

7）步骤

（1）创建引擎

（2）创建映射Base

（3）创建实体类（class）

（4）实例化

（5）创建表和删除表

（6）创建会话

3、CRUD操作

1）加add

Add添加：添加进去没提交之前可以进行更改。Session.add(self)

Session.add(self)add进去的是实例化的对象。

Session.add_all([s1,s2])

Add全部的方法。

避免主键冲突，直接设置自增性。添加的时候不指定id，主键代表的唯一的行。

try:

session.add_all([s1])

session.commit()

except:

session.rollback()

try:
    session.add_all([s1])
    session.commit()
except:
    session.rollback()

2）查query

（1）查：session.query() 实例。括号里面放入的实体类。

Query（）方法，返回的是query对象，query是可迭代对象。

（2）Session.query(Student).get(10)主键查询，返回一条传入类的实例。

3）改

改的时候必须先进行查找，查找后进行改变赋值，然后进行add加，最后要提交。

数据库操作的时候一定要进行rollback。

首先要找到这条记录。先查后改。

4）删

必须要先查找到，然后在删除。

Session.delete()

Session.commit()

4、状态

生成的每一个实体，都有一个状态属性。其类型是sqlalchemy.orm.state.lanstancestate,可以使用sqlalchemy.inspect(entity)函数查看状态。

状态	说明
Transient	实体类尚未加入到session里面，数据未保存到数据库中
Pending	Transient的实体被add()到session里面，状态切换到pending，但是还没flush到数据库中
Persistent	Session中的实体对象对应着数据库中真实的记录，pending状态在提交成功后可以变成persistent状态，或者查询成功返回的实体也是persistent状态。
Deleted	实体被删除且已经flush但未commit完成，事务提交成功了，实体变成detached，事务失败，返回persistent状态
Detached	删除成功的实体进入这个状态

新建一个实体，状态是transient临时的。

一旦add()后从transient编程pending状态。

成功commit()后从pending变成persistent()状态。

成功查询返回的实体对象，也是persistent状态。

Persistent状态的实体，修改后依然是persistent状态。

Persistent状态实体，删除后，flush后但没有commit，就变成deteled状态，成功提交，变为detached状态。

提交失败，还原到persistent状态，flush方法，主动把改变应用到数据库中去。

删除、修改操作，需要对应一个真实的记录，所以要求实体对象是persistent状态。

import sqlalchemy
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column,Integer,String,Date,Enum
from sqlalchemy.orm import sessionmaker

host = ‘192.168.118.129‘
user = ‘wang‘
password = ‘wang‘
port = 3306
database = ‘school‘

#1创建引擎
connstr = "mysql+pymysql://{}:{}@{}:{}/{}".format(user,password,host,port,database)
engine = sqlalchemy.create_engine(connstr,echo=True)

#继承base映射
Base = declarative_base()

class Student(Base): #3创建实体类
    #指定表明
    __tablename__ = ‘student‘
    id = Column(Integer,primary_key=True)
    name = Column(String(64))
    age = Column(Integer) #第一参数是字段名，如果和属性名不一致，要指定 age=Column(‘age‘,Integer)

    def __repr__(self):
        return "{}id={}name={}age={}".format(self.__class__.__name__,self.id,self.name,self.age)

#6建立会话
Session = sessionmaker(bind=engine) #返回类
session = Session()
from sqlalchemy.orm.state import InstanceState

def getstate(entity,i):
    insp = sqlalchemy.inspect(entity)
    state = ‘sessionid={},attached={} transient={},persistent={} pending={},deleted={},datched={}‘.format(
        insp.session_id,
        insp._attached,
        insp.transient,
        insp.persistent,
        insp.pending,
        insp.deleted,
        insp.detached
    )
    print(i,state)
    print(insp.key)
    print(‘=============================‘)

#4实例化
# s1 = Student(name=‘tom‘)
# print(s1.name)
# s1.age=23
# print(s1.age)
# print(s1)
#5删除创建表，少用。
# Base.metadata.drop_all(engine)
# Base.metadata.create_all(engine)

#
# #7增
# session.add(s1)
# print(s1)
# session.commit()
# print(s1)
# try:
#     session.add_all([s1])
#     session.commit()
# except:
#     session.rollback()
# #8查
students = session.query(Student).get(2)
getstate(students,1)
# for student in students:
#    print(student)
# students.name = ‘xxx‘
# students.age = ‘28‘
# session.add(students)
# session.commit()
# #9删除
# session.delete(students)
# session.commit()
try:
    students=Student(id=3,name=‘tom‘,age=23)
    getstate(students,3)
except Exception as e:
    session.rollback()
    print(e)

5、复杂查询

And：与

Or：或

not 非：

and_ or_ not_ （以后一定要加括号，因为是相当于函数调用）

in操作：

not in 取反等情况：

like ilike忽略大小写：

1）简单的条件查询

2）与 and

3）或

4）not 非

5）in 在

6）not in 不在

7）like 匹配字符串

8）ilike 匹配字符串忽略大小写

9）分页

10）排序

from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column,Integer,String,Date,Enum,FetchedValue,create_engine
from sqlalchemy.orm import sessionmaker
import enum
import sqlalchemy

Base = declarative_base()
connstr = "{}://{}:{}@{}:{}/{}".format(
    ‘mysql+pymysql‘,‘wang‘,‘wang‘,‘192.168.118.129‘,3306,
    ‘test‘
)
engine = create_engine(connstr,echo=True)
Session = sessionmaker(bind=engine)
session = Session()

class MyEnum(enum.Enum):
    M = ‘M‘
    F = ‘F‘

class Employee(Base):
    __tablename__ = ‘employees‘
    emp_no = Column(Integer,primary_key=True)
    birth_date = Column(Date,nullable=False)
    first_name = Column(String(14),nullable=False)
    last_name = Column(String(16),nullable=False)
    gender = Column(Enum(MyEnum),nullable=False)
    hire_date = Column(Date,nullable=False)

    def __repr__(self):
        return "{} no={} name={}{}gender={} ".format(
            self.__class__.__name__,
            self.emp_no,self.first_name,
            self.last_name,self.gender.value
            )
def show(emps):
    for x in emps:
        print(x)
    print(‘---------------‘,end=‘ ‘)

from sqlalchemy import not_
#简单条件查询
# emps = session.query(Employee).filter(Employee.emp_no>10015)
# show(emps)
#与and
# emps = session.query(Employee).filter(Employee.emp_no>10016).filter(Employee.emp_no<10019)
# show(emps)

#或
# emps = session.query(Employee).filter((Employee.emp_no>10018)|(Employee.emp_no<10003))
# show(emps)

# #not
# emps = session.query(Employee).filter(not_(Employee.emp_no <10018))
# show(emps)

# #in
# empslist = [10018,10019,10017,10010]
# emps = session.query(Employee).filter(Employee.emp_no.in_(empslist))
# show(emps)

# #not in
# empslist = [10018,10019,10017,10010]
# emps = session.query(Employee).filter(Employee.emp_no.notin_(empslist))
# show(emps)

# #like 匹配字符
# emps =session.query(Employee).filter(Employee.last_name.like(‘P%‘))
# show(emps)

# #ilike     忽略大小写的匹配字符
# emps = session.query(Employee).filter(Employee.last_name.ilike(‘p%‘))
# show(emps)

#分页
# emps = session.query(Employee).limit(4)
# show(emps)
#
# emps = session.query(Employee).limit(4).offset(5)
# show(emps)

#排序
#升序：
# emps = session.query(Employee).filter(Employee.emp_no>10015).order_by(Employee.emp_no)
# show(emps)

# emps = session.query(Employee).filter(Employee.emp_no>10015).order_by(Employee.emp_no.asc())
# show(emps)

#排序
#降序
# emps = session.query(Employee).filter(Employee.emp_no>10015).order_by(Employee.emp_no.desc())
# show(emps)

#多列排序
emps = session.query(Employee).filter(Employee.emp_no>10015).order_by(Employee.last_name).order_by(Employee.emp_no.desc())
show(emps)

6、消费者方法

from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column,Integer,String,Date,Enum,FetchedValue,create_engine
from sqlalchemy.orm import sessionmaker
import enum
import sqlalchemy

Base = declarative_base()
connstr = "{}://{}:{}@{}:{}/{}".format(
    ‘mysql+pymysql‘,‘wang‘,‘wang‘,‘192.168.118.129‘,3306,
    ‘test‘
)
engine = create_engine(connstr,echo=True)
Session = sessionmaker(bind=engine)
session = Session()

class MyEnum(enum.Enum):
    M = ‘M‘
    F = ‘F‘

class Employee(Base):
    __tablename__ = ‘employees‘
    emp_no = Column(Integer,primary_key=True)
    birth_date = Column(Date,nullable=False)
    first_name = Column(String(14),nullable=False)
    last_name = Column(String(16),nullable=False)
    gender = Column(Enum(MyEnum),nullable=False)
    hire_date = Column(Date,nullable=False)

    def __repr__(self):
        return "{} no={} name={}{}gender={} ".format(
            self.__class__.__name__,
            self.emp_no,self.first_name,
            self.last_name,self.gender.value
            )
def show(emps):
    for x in emps:
        print(x)
    print(‘---------------‘,end=‘ ‘)

from sqlalchemy import not_
#简单条件查询
# emps = session.query(Employee).filter(Employee.emp_no>10015)
# show(emps)
#与and
# emps = session.query(Employee).filter(Employee.emp_no>10016).filter(Employee.emp_no<10019)
# show(emps)

#或
# emps = session.query(Employee).filter((Employee.emp_no>10018)|(Employee.emp_no<10003))
# show(emps)

# #not
# emps = session.query(Employee).filter(not_(Employee.emp_no <10018))
# show(emps)

# #in
# empslist = [10018,10019,10017,10010]
# emps = session.query(Employee).filter(Employee.emp_no.in_(empslist))
# show(emps)

# #not in
# empslist = [10018,10019,10017,10010]
# emps = session.query(Employee).filter(Employee.emp_no.notin_(empslist))
# show(emps)

# #like 匹配字符
# emps =session.query(Employee).filter(Employee.last_name.like(‘P%‘))
# show(emps)

# #ilike     忽略大小写的匹配字符
# emps = session.query(Employee).filter(Employee.last_name.ilike(‘p%‘))
# show(emps)

#分页
# emps = session.query(Employee).limit(4)
# show(emps)
#
# emps = session.query(Employee).limit(4).offset(5)
# show(emps)

#排序
#升序：
# emps = session.query(Employee).filter(Employee.emp_no>10015).order_by(Employee.emp_no)
# show(emps)

# emps = session.query(Employee).filter(Employee.emp_no>10015).order_by(Employee.emp_no.asc())
# show(emps)

#排序
#降序
# emps = session.query(Employee).filter(Employee.emp_no>10015).order_by(Employee.emp_no.desc())
# show(emps)

#多列排序
# emps = session.query(Employee).filter(Employee.emp_no>10015).order_by(Employee.last_name).order_by(Employee.emp_no.desc())
# show(emps)

#总行数
emps = session.query(Employee)
# print(len(list(emps))) #返回大量的结果集，然后转换list
# print(emps.count())   #聚合函数count(*)的查询

#取所有数据
# print(emps.all())

#取行首
# print(emps.first())

#有且只有一行
#print(emps.one())#会抛出异常
print(emps.limit(1).one())

#删除 delete by query
session.query(Employee).filter(Employee.emp_no > 10018).delete()
#session.commit() 提交后删除

查找等情况：

只是简单查找count的结果还是很少的。

7、聚合函数、分组查询

1）count

2）max/min/avg

3）分组

Scalar：有且只有一行的第一个元素。

from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column,Integer,String,Date,Enum,FetchedValue,create_engine
from sqlalchemy.orm import sessionmaker
import enum
import sqlalchemy

Base = declarative_base()
connstr = "{}://{}:{}@{}:{}/{}".format(
    ‘mysql+pymysql‘,‘wang‘,‘wang‘,‘192.168.118.129‘,3306,
    ‘test‘
)
engine = create_engine(connstr,echo=True)
Session = sessionmaker(bind=engine)
session = Session()

class MyEnum(enum.Enum):
    M = ‘M‘
    F = ‘F‘

class Employee(Base):
    __tablename__ = ‘employees‘
    emp_no = Column(Integer,primary_key=True)
    birth_date = Column(Date,nullable=False)
    first_name = Column(String(14),nullable=False)
    last_name = Column(String(16),nullable=False)
    gender = Column(Enum(MyEnum),nullable=False)
    hire_date = Column(Date,nullable=False)

    def __repr__(self):
        return "{} no={} name={}{}gender={} ".format(
            self.__class__.__name__,
            self.emp_no,self.first_name,
            self.last_name,self.gender.value
            )
def show(emps):
    for x in emps:
        print(x)
    print(‘---------------‘,end=‘ ‘)

from sqlalchemy import not_
#简单条件查询
# emps = session.query(Employee).filter(Employee.emp_no>10015)
# show(emps)
#与and
# emps = session.query(Employee).filter(Employee.emp_no>10016).filter(Employee.emp_no<10019)
# show(emps)

#或
# emps = session.query(Employee).filter((Employee.emp_no>10018)|(Employee.emp_no<10003))
# show(emps)

# #not
# emps = session.query(Employee).filter(not_(Employee.emp_no <10018))
# show(emps)

# #in
# empslist = [10018,10019,10017,10010]
# emps = session.query(Employee).filter(Employee.emp_no.in_(empslist))
# show(emps)

# #not in
# empslist = [10018,10019,10017,10010]
# emps = session.query(Employee).filter(Employee.emp_no.notin_(empslist))
# show(emps)

# #like 匹配字符
# emps =session.query(Employee).filter(Employee.last_name.like(‘P%‘))
# show(emps)

# #ilike     忽略大小写的匹配字符
# emps = session.query(Employee).filter(Employee.last_name.ilike(‘p%‘))
# show(emps)

#分页
# emps = session.query(Employee).limit(4)
# show(emps)
#
# emps = session.query(Employee).limit(4).offset(5)
# show(emps)

#排序
#升序：
# emps = session.query(Employee).filter(Employee.emp_no>10015).order_by(Employee.emp_no)
# show(emps)

# emps = session.query(Employee).filter(Employee.emp_no>10015).order_by(Employee.emp_no.asc())
# show(emps)

#排序
#降序
# emps = session.query(Employee).filter(Employee.emp_no>10015).order_by(Employee.emp_no.desc())
# show(emps)

#多列排序
# emps = session.query(Employee).filter(Employee.emp_no>10015).order_by(Employee.last_name).order_by(Employee.emp_no.desc())
# show(emps)

#总行数
# emps = session.query(Employee)
# # print(len(list(emps))) #返回大量的结果集，然后转换list
# # print(emps.count())   #聚合函数count(*)的查询
#
# #取所有数据
# # print(emps.all())
#
# #取行首
# # print(emps.first())
#
# #有且只有一行
# #print(emps.one())#会抛出异常
# print(emps.limit(1).one())
#
# #删除 delete by query
# session.query(Employee).filter(Employee.emp_no > 10018).delete()
# #session.commit() 提交后删除

#聚合函数
from sqlalchemy import func

query = session.query(func.count(Employee.emp_no))
# print(query.one()) 返回的是二元组
# print(query.scalar())
#
#
# #max，min，avg
# print(session.query(func.max(Employee.emp_no)).scalar())
# print(session.query(func.min(Employee.emp_no)).scalar())
# print(session.query(func.avg(Employee.emp_no)).scalar())

#分组
print(session.query(Employee.gender,func.count(Employee.emp_no)).group_by(Employee).all())

8、关联查询

级联删除，就是主键删除，外键也是要删除的。

不删除和组织删除等。

from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column,Integer,String,Date,Enum,FetchedValue,create_engine,ForeignKey
from sqlalchemy.orm import sessionmaker
import enum
import sqlalchemy

Base = declarative_base()
connstr = "{}://{}:{}@{}:{}/{}".format(
    ‘mysql+pymysql‘,‘wang‘,‘wang‘,‘192.168.118.129‘,3306,
    ‘test‘
)
engine = create_engine(connstr,echo=True)
Session = sessionmaker(bind=engine)
session = Session()

class MyEnum(enum.Enum):
    M = ‘M‘
    F = ‘F‘

class Employee(Base):
    __tablename__ = ‘employees‘
    emp_no = Column(Integer,primary_key=True)
    birth_date = Column(Date,nullable=False)
    first_name = Column(String(14),nullable=False)
    last_name = Column(String(16),nullable=False)
    gender = Column(Enum(MyEnum),nullable=False)
    hire_date = Column(Date,nullable=False)

    def __repr__(self):
        return "{} no={} name={}{}gender={} ".format(
            self.__class__.__name__,
            self.emp_no,self.first_name,
            self.last_name,self.gender.value
            )

class Department(Base):
    __tablename__ = ‘departments‘

    dept_no = Column(String(4),primary_key=True)
    dept_name = Column(String(40),nullable=False,unique=True)

    def __repr__(self):
        return "{} dept_no={} dept_name={}".format(
            self.__name__,
            self.dept_no,self.dept_name
            )

class Dept_emp(Base):
    __tablename__ = ‘dept_emp‘
    emp_no = Column(Integer,ForeignKey(‘employess.emp_no‘,ondelete=‘CASCADE‘),primary_key=True)
    dept_no = Column(String(4),ForeignKey(‘departments.dept_no‘,ondelete=‘CASCADE‘),primary_key=True)
    from_date = Column(Date,nullable=False)
    to_date = Column(Date,nullable=False)

    def __repr__(self):
        return ‘{}empno={} deptno={}‘.format(type(self).__name__,self.emp_no,self.dept_no)

#ForeignKey(‘employess.emp_no‘,ondelete=‘CASCADE‘)定义外键约束。

10、总结

开发中，一般都会采用orm框架，这样可以使用对象操作表了。

定义表映射的类，使用column的描述器定义类属性，使用foreignkey来定义外键约束，。

在一个对象中，想查看其它表对应的对象的内容，就使用relationship来定义关系。

使用外键能够保证数据完整性一致性：

使用外键的话开发难度大，大数据的时候影响插入、修改、删除的效率。

在业务层保证数据的一致性。

以上是关于SQLAlchemy的使用的主要内容，如果未能解决你的问题，请参考以下文章

SQLAlchemy使用笔记－－SQLAlchemy ORM

flask_sqlalchemy和sqlalchemy联系区别及其使用方式

如何在 SQLAlchemy 模型中使用 SQLAlchemy Utils

sqlalchemy使用之查询指定的列

Flask-SQLAlchemy 和 SQLAlchemy

python web开发-flask中使用sqlalchemy