SQLAlchemy基础
SQLAlchemy概述 SQLAlchemy安装 SQLAIchemy简介 SQLAIchemy构架 ORM模型
数据库对象管理 连接mysql 声明映射 创建映射类 创建构架 创建映射类的实例 创建会话类 添加新对象 外键约束
SQLAlchemy进阶
查询操作 基本查询 使用ORM描述符进行查询 使用命名元组 修改显示字段名 排序 提取部分数据 结果过滤 常用过滤操作符 查询对象返回值 聚合 多表查询
修改操作 更新数据 删除记录
能兼容各种数据库，但要结合方言
-----------------------
from sqlalchemy import create_engine, Column, Integer, String, ForeignKey, Date
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker #创建会话类
engine = create_engine( #创建mysql引擎
'mysql+pymysql://root:tedu.cn@localhost/tedu1807?charset=utf8',
encoding='utf8',
echo=True #表示将日志输出到终端屏幕，默认为False
)
Session = sessionmaker(bind=engine)
Base = declarative_base()
class Departments(Base):
__tablename__ = 'departments'
dep_id = Column(Integer, primary_key=True)
dep_name = Column(String(20), unique=True, nullable=False)
def __str__(self):
return '部门： %s' % self.dep_name
class Employees(Base):
__tablename__ = 'employees'
emp_id = Column(Integer, primary_key=True)
emp_name = Column(String(20))
gendar = Column(String(6))
email = Column(String(50))
phone = Column(String(11))
dep_id = Column(Integer, ForeignKey('departments.dep_id') )
def __str__(self):
return '员工： %s' % self.emp_name
class Salary(Base):
__tablename__ = 'salary'
auto_id = Column(Integer, primary_key=True) # auto_id
date = Column(Date)
emp_id = Column(Integer, ForeignKey('employees.emp_id')) #不用reference
basic = Column(Integer)
awards = Column(Integer)
def __str__(self):
return '工资：%s:%s->%s' % (self.date, self.emp_id, (self.basic+self.awards))
if __name__ == '__main__':
Base.metadata.create_all(engine)
------------------------------insert/add
from dbconn import Departments, Session, Employees, Salary
hr = Departments(dep_id=1, dep_name='hr')
op = Departments(dep_id=2, dep_name='运维部')
dev = Departments(dep_id=3, dep_name='开发部')
qa = Departments(dep_id=4, dep_name='测试')
session = Session()
session.add(hr)
session.add_all([op, dev, qa])
fa = Employees(emp_id=1, emp_name='xa', gendar='na', email='ss@xxx', phone='1501756890', dep_id=1)
fb = Employees(emp_id=2, emp_name='xb', gendar='na', email='ss@xxx', phone='1501759890', dep_id=4)
fc = Employees(emp_id=3, emp_name='xc', gendar='na', email='ss@xxx', phone='1517596890', dep_id=3)
fd = Employees(emp_id=4, emp_name='xd', gendar='na', email='ss@xxx', phone='1501796890', dep_id=1)
ff = Employees(emp_id=5, emp_name='xxf', gendar='na', email='ss@xxx', phone='1501759890', dep_id=2)
fg = Employees(emp_id=6, emp_name='xxg', gendar='na', email='ss@xxx', phone='1017596890', dep_id=4)
fh = Employees(emp_id=7, emp_name='xxh', gendar='na', email='ss@xxx', phone='1507596890', dep_id=1)
fi = Employees(emp_id=8, emp_name='xxi', gendar='na', email='ss@xxx', phone='1501759680', dep_id=2)
fj = Employees(emp_id=9, emp_name='xxj', gendar='na', email='ss@xxx', phone='1501596890', dep_id=4)
fk = Employees(emp_id=10, emp_name='xxk', gendar='na', email='ss@xxx', phone='1501796890', dep_id=1)
session.add_all([fa, fb, fc, fd, ff, fg, fh, fi, fj, fk])
session.commit()
session.close()
------------------------------insert/add
------------------------------select/查询
from dbconn import Departments, Session, Employees, Salary
session = Session()
##############查询
query1 = session.query(Departments)
print(query1) # query1只是sql语句
print(list(query1)) #列表是所有实例的集合
for dep in query1:
print('%s : %s' % (dep.dep_id, dep.dep_name))
#############查询，直接用list,结果是列表
query2 = session.query(Employees.emp_id, Employees.emp_name)
print(query2)
print(list(query2))
############排序
query3 = session.query(Departments).order_by(Departments.dep_id)
print(query3)
for dep in query3:
print('%s : %s' % (dep.dep_id, dep.dep_name))
#############切片
query4 = session.query(Departments).order_by(Departments.dep_id)[0:2]
print(query4) # 取切片后，就不是sql语句了，而是实例集合
for dep in query4:
print('%s : %s' % (dep.dep_id, dep.dep_name))
#############filter过滤
query5 = session.query(Departments).filter(Departments.dep_id==3) # 记得要加过滤的条件
print(query5)
for dep in query5:
print('%s : %s' % (dep.dep_id, dep.dep_name))
############filter多次过滤
query6 = session.query(Employees).filter(Employees.dep_id==2)\
.filter(Employees.email=='ss@xxx')
print(query6)
for emp in query6:
print('%s : %s' % (emp.dep_id, emp.emp_name))
############filter 过滤符：in
query7 = session.query(Departments).filter(Departments.dep_id.in_([1,2,5]))
print(query7)
for dep in query7:
print('%s: %s' % (dep.dep_id, dep.dep_name))
########### 查询对象返回值
query1 = session.query(Departments)
print(query1.all()) # 取出全部，构成实例集合
query5 = session.query(Departments).filter(Departments.dep_id==3)
print(query5.one()) # 返回一个具体的实例
query8 = session.query(Departments.dep_name, Departments.dep_id).filter(Departments.dep_id==2)
print(query8.scalar()) # 调用one, 返回结果的第一项
# query9 = session.query(Departments.dep_id,Departments.dep_name).filter(Departments.dep_id==2)
# print(query9.scalar())
############聚合(算合)
query9 = session.query(Departments).count()
print(query9)
############ 多表查询
query10 = session.query(Employees.emp_name, Departments.dep_name)\
.join(Departments, Employees.dep_id==Departments.dep_id)
print(query10.all())
query11 = session.query(Departments.dep_name, Employees.emp_name)\
.join(Employees, Employees.dep_id==Departments.dep_id)
print(query11.all())
session.close()
------------------------------select/查询
------------------------------更新数据， 删除数据
from dbconn import Departments, Session, Employees
session = Session()
############# 更新数据
modify1 = session.query(Departments).filter(Departments.dep_name=='hr')
hr = modify1.one()
print(hr)
hr.dep_name = '人力资源部' # 通过会话的字段赋值更新
##############session.delete()
delete = session.query(Employees).filter(Employees.emp_id==10)
emp = delete.one()
session.delete(emp)
session.commit()
session.close()
------------------------------
>>> sqlalchemy.
Display all 130 possibilities? (y or n)
sqlalchemy.ARRAY( sqlalchemy.asc(
sqlalchemy.BIGINT( sqlalchemy.between(
sqlalchemy.BINARY( sqlalchemy.bindparam(
sqlalchemy.BLANK_SCHEMA sqlalchemy.case(
sqlalchemy.BLOB( sqlalchemy.cast(
sqlalchemy.BOOLEAN( sqlalchemy.collate(
sqlalchemy.BigInteger( sqlalchemy.column(
sqlalchemy.Binary( sqlalchemy.cprocessors
sqlalchemy.Boolean( sqlalchemy.create_engine(
sqlalchemy.CHAR( sqlalchemy.cresultproxy
sqlalchemy.CLOB( sqlalchemy.cutils
sqlalchemy.CheckConstraint( sqlalchemy.delete(
sqlalchemy.Column( sqlalchemy.desc(
sqlalchemy.ColumnDefault( sqlalchemy.dialects
sqlalchemy.Constraint( sqlalchemy.distinct(
sqlalchemy.DATE( sqlalchemy.engine
sqlalchemy.DATETIME( sqlalchemy.engine_from_config(
sqlalchemy.DDL( sqlalchemy.event
sqlalchemy.DECIMAL( sqlalchemy.events
sqlalchemy.Date( sqlalchemy.exc
sqlalchemy.DateTime( sqlalchemy.except_(
sqlalchemy.DefaultClause( sqlalchemy.except_all(
sqlalchemy.Enum( sqlalchemy.exists(
sqlalchemy.FLOAT( sqlalchemy.extract(
sqlalchemy.FetchedValue( sqlalchemy.false(
sqlalchemy.Float( sqlalchemy.func(
sqlalchemy.ForeignKey( sqlalchemy.funcfilter(
sqlalchemy.ForeignKeyConstraint( sqlalchemy.insert(
sqlalchemy.INT( sqlalchemy.inspect(
sqlalchemy.INTEGER( sqlalchemy.inspection
sqlalchemy.Index( sqlalchemy.interfaces
sqlalchemy.Integer( sqlalchemy.intersect(
sqlalchemy.Interval( sqlalchemy.intersect_all(
sqlalchemy.JSON( sqlalchemy.join(
sqlalchemy.LargeBinary( sqlalchemy.lateral(
sqlalchemy.MetaData( sqlalchemy.literal(
sqlalchemy.NCHAR( sqlalchemy.literal_column(
sqlalchemy.NUMERIC( sqlalchemy.log
sqlalchemy.NVARCHAR( sqlalchemy.modifier(
sqlalchemy.Numeric( sqlalchemy.not_(
sqlalchemy.PassiveDefault( sqlalchemy.null(
sqlalchemy.PickleType( sqlalchemy.nullsfirst(
sqlalchemy.PrimaryKeyConstraint( sqlalchemy.nullslast(
sqlalchemy.REAL( sqlalchemy.or_(
sqlalchemy.SMALLINT( sqlalchemy.outerjoin(
sqlalchemy.Sequence( sqlalchemy.outparam(
sqlalchemy.SmallInteger( sqlalchemy.over(
sqlalchemy.String( sqlalchemy.pool
sqlalchemy.TEXT( sqlalchemy.processors
sqlalchemy.TIME( sqlalchemy.schema
sqlalchemy.TIMESTAMP( sqlalchemy.select(
sqlalchemy.Table( sqlalchemy.sql
sqlalchemy.Text( sqlalchemy.subquery(
sqlalchemy.ThreadLocalMetaData( sqlalchemy.table(
sqlalchemy.Time( sqlalchemy.tablesample(
sqlalchemy.TypeDecorator( sqlalchemy.text(
sqlalchemy.Unicode( sqlalchemy.true(
sqlalchemy.UnicodeText( sqlalchemy.tuple_(
sqlalchemy.UniqueConstraint( sqlalchemy.type_coerce(
sqlalchemy.VARBINARY( sqlalchemy.types
sqlalchemy.VARCHAR( sqlalchemy.union(
sqlalchemy.alias( sqlalchemy.union_all(
sqlalchemy.all_( sqlalchemy.update(
sqlalchemy.and_( sqlalchemy.util
sqlalchemy.any_( sqlalchemy.within_group(
网友评论