标签：连接方式 prim ada pad query 复杂查询 avg for stat
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:wang@192.168.118.145:3306/mysql”)
⑵pymysql
Mysql+pymysql://<user>:<password>@<host>/<dbname>[?<options>]
engine=sqlalchemy.create_engine(“mysql+pymysql:wang:wang@192.168.118.145:3306/test”)
engine= sqlalchemy.create_engine(“mysql+pymysql:wang:wang@192.168.118.145: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都是线程安全的，有一个就行了。
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.145‘
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)
#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)
#6建立会话
Session = sessionmaker(bind=engine)
#返回类
session = Session()
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()
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.145‘
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)
#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)
#6建立会话
Session = sessionmaker(bind=engine)
#返回类
session = Session()
session.add(s1)
print(s1)
session.commit()
print(s1)
try:
session.add_all([s1])
session.commit()
except:
session.rollback()
2）查query
（1）查：session.query() 实例。括号里面放入的实体类。
Query（）方法，返回的是query对象，query是可迭代对象。
（2）Session.query(Student).get(10)主键查询，返回一条传入类的实例。
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.128‘
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)
#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)
#6建立会话
Session = sessionmaker(bind=engine)
#返回类
session = Session()
session.add(s1)
print(s1)
session.commit()
print(s1)
try:
session.add_all([s1])
session.commit()
except:
session.rollback()
students = session.query(Student)
for student in students:
print(student)
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.128‘
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)
#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)
#6建立会话
Session = sessionmaker(bind=engine)
#返回类
session = Session()
session.add(s1)
print(s1)
session.commit()
print(s1)
try:
session.add_all([s1])
session.commit()
except:
session.rollback()
students = session.query(Student).get(2)
# for student
in students:
# print(student)
print(students)
3）改
改的时候必须先进行查找，查找后进行改变赋值，然后进行add加，最后要提交。
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)
#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)
#6建立会话
Session = sessionmaker(bind=engine)
#返回类
session = Session()
#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)
# for student
in students:
# print(student)
students.name
= ‘xxx‘
students.age
= ‘28‘
session.add(students)
session.commit()
数据库操作的时候一定要进行rollback。
首先要找到这条记录。先查后改。
4）删
必须要先查找到，然后在删除。
Session.delete()
Session.commit()
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)
#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)
#6建立会话
Session = sessionmaker(bind=engine)
#返回类
session = Session()
#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)
# for student
in students:
# print(student)
# students.name = ‘xxx‘
# students.age = ‘28‘
# session.add(students)
# session.commit()
#9删除
session.delete(students)
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={}\ntransient={},persistent={}\npending={},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、复杂查询
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=‘\r\n‘)
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))
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=‘\r\n‘)
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=‘\r\n‘)
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=‘\r\n‘)
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来定义关系。
使用外键能够保证数据完整性一致性：
使用外键的话开发难度大，大数据的时候影响插入、修改、删除的效率。
在业务层保证数据的一致性。
标签：连接方式 prim ada pad query 复杂查询 avg for stat
原文地址：https://www.cnblogs.com/wangchunli-blogs/p/9949927.html
