python 模块 SQLalchemy

 SQLalchemy 概述:

# &&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&
# SQLAlchemy ORM框架(需要依赖数据库API语言)

# 架构图
# SQLalchemy orm
# SQLalchemy core
#       --> schema/types 架构/类型
#       --> sql expression language SQL表达式语言
#       --> engine(框架引擎)
#           --> connection pooling (数据库连接池)
#           --> diaiect (选择连接数据路的DB api种类)
# DBAPI(pymysql,oracle,sqlite)


执行流程:
    类/对箱操作 -> SQL -> pymsql、mysqldb -> 在去数据中执行

连接数据库格式:

连接数据库:
    MYSQL - python
    mysql+mysqldb://<user>:<password>@<host>:<port>/<dbname>
    
    pymysql
    mysql+pymsql://<username>:<password>@<host>/<dbname>
    示例:
    "mysql+pymysql://root:123456@127.0.0.1:3306/t1?charset=utf8"
   
    cx_Oracle
    oracle+cx_oracle://user:pwd@host:port/dbname 

线程池/单例方式

from sqlalchemy import create_engine
from sqlalchemy.orm import Session,scoped_session,sessionmaker
# 创建连接
# engine = create_engine("mysql+pymysql://root:123456@127.0.0.1:3306/anec?charset=utf8")

# # 这种方式自动从连接池中获取连接(方式一)
sess = sessionmaker(bind=engine)
s = scoped_session(sess)

# 单例方式(方式二)
s = Session(bind=engine)

创建数据表  常用数据类型

from datetime import datetime
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import (create_engine, Column, Integer, String, SmallInteger, DateTime,Float,DECIMAL,Boolean,Enum,Date,Time,Text)
from sqlalchemy.dialects.mysql import LONGTEXT

engine = create_engine("mysql+pymysql://root:QWer1234!!@172.16.3.108:3306/cmdb?charset=utf8")
Base = declarative_base()

class Human2(Base):
    __tablename__ = "human2"
    id = Column("id", Integer, autoincrement=True, primary_key=True)    
　　 # mysql 为 int(11)
    name = Column("name", String(20), nullable=False, unique=True)      
　　 # mysql 为 varchar(20)
    age = Column("age", Integer, nullable=False)
    sex = Column("sex", SmallInteger, default=1)
    price = Column('price',Float,nullable=True)                         
　　 # mysql 为 float 保留3位小数
    price2 = Column(DECIMAL(7, 3))                                      
　　 # float精度类型 mysql 为 decimal(7,3)
    delete = Column(Boolean)                                            
　　 # mysql 为 tinyint(1)  值为 1/0
    sex2 = Column(Enum("男", "女"))                                      
　　 # mysql 为 enum("男", "女")
    create_time1 = Column("create_time", DateTime, default=datetime.now) 
　　 # mysql 为 datetime()
    create_time2 = Column(Date)                                         
　　 # 只能存储指定的年月日 mysql 为 date
    create_time3 = Column(Time)                                         
　　 # 只能存储指定的时间 mysql 为 time
    content = Column(Text)                                              
　　 #  mysql 为 text
    content2 = Column(LONGTEXT)                                         
　　 # 注意从sqlalchemy.dialects.mysql 导入,longtext

    def __repr__(self):
        return "name {}".format(self.name)

# 执行在数据库中创建表
Base.metadata.create_all(bind=engine)

四种外键约束

from sqlalchemy import ForeignKey

四种外键约束:
1.第一种：RESTRICT(默认就是这种。当父表数据被删除，从表会拒绝删除)
语法：uid = Column(Integer , ForeignKey("user.id" ,ondelete="RESTRICT")) 

2.第二种：NO ACTIION（同RESTRICT 一样）
语法: uid = Column(Integer , ForeignKey("user.id" ,ondelete="NO ACTION")) 

3.第三种：CASCADE (父表数据删除、从表数据也会跟着删除)
语法：uid = Column(Integer , ForeignKey("user.id" ,ondelete="CASCADE"))

4.第四种： SET NULL (父表数据删除，从表外键字段设为NULL)
语法：uid = Column(Integer , ForeignKey("user.id" ,ondelete="SET NULL"))
注意： 如果uid字段设置了 nullable=False , 再设置 ondelete = "SET NULL",pycharm运行程序则会报错

经典样式 创建数据表

# 经典样式创建表
from sqlalchemy import (Table, MetaData, create_engine,Column, Integer, String, SmallInteger, DateTime)
from datetime import datetime
from sqlalchemy.orm import mapper

engine = create_engine("mysql+pymysql://root:QWer1234!!@172.16.3.108:3306/cmdb?charset=utf8")
metadata = MetaData()

# 创建 user 表
user = Table("user", metadata,
        Column("id", Integer, nullable=False, primary_key=True, autoincrement=True),
        Column("username", String(20), nullable=False),
        Column("age", Integer, nullable=False),
        Column("sex", SmallInteger, default=1),
        Column("create_time", DateTime, default=datetime.now)
    )

# model
class User(object):
    def __init__(self, username=None, age=None, sex=None):
        if username:
            self.username = username
        if age:
            self.age =age
        if sex:
            self.sex =sex

# table与model映射
mapper(User, user)

# 执行在数据库中创建表
metadata.create_all(bind=engine)

经典样式 操作表

from sqlalchemy.orm import  sessionmaker
from sqlalchemy  import create_engine


engine = create_engine("mysql+pymysql://root:QWer1234!!@172.16.3.108:3306/cmdb?charset=utf8")
Session = sessionmaker(bind=engine)
s = Session()

# 示例话 创建的表类(当前没有的话,需要导入创建表)
from 导入经典样式的表类 import User
try:
    user = User("rose", 20, 0)
    s.add(user)
    s.commit()
except Exception as e:
    s.rollback()

ORM样式 创建数据表

# orm样式创建表
from datetime import datetime
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import (create_engine, Column, Integer, String, SmallInteger, DateTime)
from sqlalchemy.orm import Session

engine = create_engine("mysql+pymysql://root:QWer1234!!@172.16.3.108:3306/cmdb?charset=utf8")
Base = declarative_base()

class Human(Base):
    __tablename__ = "human"
    id = Column("id", Integer, autoincrement=True, primary_key=True)
    name = Column("name", String(20), nullable=False, unique=True)
    age = Column("age", Integer, nullable=False)
    sex = Column("sex", SmallInteger, default=1)
    create_time = Column("create_time", DateTime, default=datetime.now)

    def __repr__(self):
        return "name {}".format(self.name)

# 执行在数据库中创建表
Base.metadata.create_all(bind=engine)

Base.metadata.drop_all(engine)

s = Session(bind=engine)

# 实例化创建表
h = Human(name="king002", age=30, sex=1)

# 添加数据至human表中
s.add(h)
s.commit()

relationship关联创建表

from sqlalchemy import create_engine
from sqlalchemy import Column, Integer, ForeignKey,String,DateTime
from sqlalchemy.orm import relationship
from sqlalchemy.ext.declarative import declarative_base
import datetime


engine = create_engine("mysql+pymysql://root:123456@127.0.0.1:3306/anec?charset=utf8")

Base = declarative_base()
class Parent(Base):
    __tablename__ = 'parent'
    id = Column(Integer, primary_key=True,autoincrement=True)
    name = Column(String(64))
    ctime = Column(DateTime,default=datetime.datetime.now)
    children = relationship("Child")

    def __repr__(self):
        return "name {}".format(self.name)
class Child(Base):
    __tablename__ = 'child'
    id = Column(Integer, primary_key=True,autoincrement=True)
    name = Column(String(64))
    parent_id = Column(Integer, ForeignKey('parent.id'))

    def __repr__(self):
        return "name {}".format(self.name)

# Base.metadata.create_all(engine)
# Base.metadata.drop_all(engine)

数据库中已存在数据表,反射连接数据库中的表

方式一:

from sqlalchemy import create_engine
from sqlalchemy import MetaData
from sqlalchemy import select

# 连接数据库
engine = create_engine("mysql+pymysql://root:QWer1234!!@172.16.3.108:3306/cmdb?charset=utf8", pool_size=5)

# 从连接的数据库池中获取一个连接
conn = engine.connect()

# 实例化一个 表的类
metadata = MetaData()

# 指定反射与连接数据库绑定
metadata.reflect(bind=engine)

# 获取数据库中的表对象:(这里表名是 t1 )
# 方式一:
t1 = metadata.tables.get('t1')

# 方式二:
from sqlalchemy import Table
t2 = Table('t1', metadata, autoload=True, autoload_with=engine)

# 连接执行查询,返回查询内容( t1.c.字段名 )
info = conn.execute(select([t1.c.id,t1.c.name]).limit(3)).fetchall()
print(info)

cc = select([t1.c.id,t1.c.name]).where(t1.c.id>8).limit(3).offset(0)
info2 = conn.execute(cc).fetchall()
print(info2)

# cc = select([t1.c.it, func.count(t1.c.id), func.sum(t1.c.age)]).group_by(human.c.sex)
# info2 = conn.execute(cc).fetchall()
# print(info2)

# 连表查询
# i = requirement.join(project, project.c.id==requirement.c.prj_id).join(product, product.c.id==project.c.prod_id)
# s = select([project.c.prj_name.label("prj_name"), product.c.prod_name.label("prod_name"), requirement.c.req_name.label("req_name")]).select_from(i)
# res = conn.execute(s).fetchall()

# 插入数据:
#   插入字典数据:
tt1 = t1.insert()
conn.execute(tt1,[{'name':'con6'},{'name':'con5'}])
conn.close()

# 指定字段插入:
tt2 = t1.insert().values(name='con6')
conn.execute(tt2)
conn.close()

# 修改数据:
tt3 = t1.update().values(name='ccccc').where(t1.c.id == 1)
conn.execute(tt3)
conn.close()

# 删除数据:
tt4 = t1.delete().where(t1.c.id == 6)
conn.execute(tt4)
conn.close()

方式二:

# ORM样式反射存在的表 返回sqlalchemy  sqlalchemy.ext.automap 对象:
from sqlalchemy.ext.automap import automap_base
from sqlalchemy import create_engine
from sqlalchemy.orm import Session
from datetime import datetime

engine = create_engine("mysql+pymysql://root:QWer1234!!@172.16.3.108:3306/cmdb?charset=utf8")

Base = automap_base()
Base.prepare(engine, reflect=True)

# 反射得到orm
Human = Base.classes.t1

# 通信
session = Session(bind=engine)

# 插入数据:
# h = Human(name="vcr")
# session.add(h)
# # session.add_all(h)
# session.commit()

# # 修改数据
# h_obj = session.query(Human).filter_by(name="vcr").first()
# h_obj.name = "aaaaaaaaaaaaaaaaaaa"
# session.add(h_obj)
# session.commit()

# # 删除数据
# h_obj = session.query(Human).filter_by(name="aaaaaaaaaaaaaaaaaaa").first()
# session.delete(h_obj)
# session.commit()

# 查询数据(为sqlalchemy.ext.automap 对象)
# res = session.query(Human).filter(Human.id > 7)
# for i in res:
#     print(i.name)

数据操作:

######################################## 查
# 查询所有数据 [ obj,obj2,.... ]
# info = s.query(Parent).all()
# for i in info:
#     print(i.name)

# # 参数查询
# info = s.query(Parent).filter_by(id=2).
# for i in info:
#     print(i.id)
#     print(i.name)
#     print(i.ctime)

# 语句查询
# from sqlalchemy import text
# info = s.query(Parent).from_statement(text("select * from Parent where id = 2")).all()
# print(info)

# 条件查询
# info = s.query(Parent).filter(Parent.id >2)
# for i in info:
#     print(i.name)
#     print(i.ctime)
# ret = s.query(Users).filter_by(name='alex').all()
# ret = s.query(Users).filter(Users.id > 1,Users.name == 'alex').all()
# ret = s.query(Users).filter(Users.id.between(1,3),Users.name == 'alex').all()
# ret = s.query(Users).filter(Users.id.in_([1,3,4])).all()
# ret = s.query(Users).filter(~Users.id.in_([1,3,4])).all()
# ret = s.query(Users).filter(Users.id.in_(s.query(Users.id).filter_by(name='alex')))
#
# from sqlalchemy import and_,or_
#
# ret = s.query(Users).filter(and_(Users.id > 3,User.name == 'alex')).all()
# ret = s.query(Users).filter(or_(Users.id > 3,User.name == 'alex')).all()
# ret = s.query(Users).filter(
#     or_(
#         Users.id > 3,
#         and_(User.name == 'alex',Users.id>3),
#         Users.extra != ''
#     )).all()

# # 通配符
# ret= s.query(User).filter(User.name.like('e%')).all()
# ret= s.query(User).filter(~User.name.like('e%')).all()

# 限制
# ret = s.query(Users)[1:3]

# # 排序
# ret = s.query(Users).order_by(Users.name.desc()).all()
# ret = s.query(Users).order_by(Users.name.desc(),User.id.asc()).all()

# # 分组
# # from sqlalchemy.sql import func
# ret = s.query(Users).group_by(Users.extra).all()
# ret = s.query(
#     func.max(Users.id),
#     func.sum(Users.id),
#     func.min(Users.id)).group_by(Users.name).all()
# ret = s.query(
#     func.max(Users.id),
#     func.sum(Users.id),
#     func.min(Users.id)).group_by(Users.name).having(func.min(Users.id)>2).all()

# # 连表
# ret = s.query(Users,Favor).filter(Users.id == Favor.user_id).all()
# ret = s.query(Users).join(Favor).all()
# ret = s.query(Users).join(Favor,isouter=True).all()

# # 组合
# q1 = s.query(Users.name).filter(Users.id > 2)
# q2 = s.query(Favor.caption).filter(Favor.nid < 2)
# ret = q1.union(q2).all()
#
# ret = q1.union_all(q2).all()

######################################## 增
# # 添加数据至human表中
# s.add(h)        # 数据格式为 {}
# s.add_all(h)   数据格式为:  [{},{}...]
# s.commit()

# ----------------------------------
    # sqlalchemy 连表,一对多连表查询
    # 方式一(连表查询方式)
    # s.query(表1,表2).join(表2,isouter=True).all()

    # 方式二,在创建表的时候加入关联字段
    #   外键所在的表增加如下一个字段:
    #   字段名 = relationship("外连表",backref='pers(反向字段)')
    #   backref='pers' 反向关联

    # 添加以后查询方式:
    # info = s.query(表1).all()
    # for i in info:
    #     print(表1.字段,表1.关联字段名.表2)

    # # sqlalchemy跨表增加
    # 方式一
    # hb = 表1(参数)
    # hb.pers = [表2(参数),表2(参数)]
    # s.add(hb)
    # s.commit()

    # 方式二
    # hb = 表1(name=参数,关联字段=表2(参数))
    # s.add(hb)
    # s.commit()

# ----------------------------------
    # sqlalchemy 连表,多对多连表查询
    # 男孩表
    # 男女孩约会表
    # 女孩表

    # 添加查询字段:
    # 在男孩表中添加:
    # girls = relationship('女孩表',secondary='男女孩约会表',backref='男孩表')
    # 要在女孩表中也可以添加
    # 或者在约会表中也可以添加两个字段

######################################## 删
# 删除
# s.query(Parent).filter(Parent.id == 4).delete()
# s.commit()

######################################## 修
# 修改
# s.query(Parent).filter(Parent.id == 3).update({'name':'aaaaaaaa'})
# s.query(Parent).filter(Parent.id == 3).update({Parent.name:Parent.name +'傻逼'},synchronize_session=False)  # 字符互操作
# s.query(Parent).filter(Parent.id == 3).update({Parent.age:Parent.name + 8 },synchronize_session='evaluate')  # 数字互操作
# s.commit()

数据库反向生成sqlalchemy ORM

# 数据表反向生成sqlalchemy ORM:
# 依赖sqlacodegen 模块
# 应用:
# sqlacodegen --tables student(表名) --outfile ./a.py(输出的文件名) mysql+pymysql://数据库用户名:密码@地址/数据库?charset=utf8
# windows 环境下cmd窗口 执行 即可.(注意输出的文件)

原生sql语句执行:

 # sqlalchemy 原生sql语句应用:

    # 1.用text包含sql语句,也可以传递参数至sql语句中
    # from sqlalchemy import text
    # result = db.execute(text('select * from table where id < :id and typeName=:type'), {'id': 2, 'type': 'USER_TABLE'})

    # 2.如果不指定parameter的类型, 默认为字符串类型;
    # 如果要传日期参数, 需要使用text()
    # 的bindparams参数来声明
    # from sqlalchemy import DateTime
    # date_param = datetime.today() + timedelta(days=-1 * 10)
    # sql = "delete from caw_job_alarm_log  where alarm_time<:alarm_time_param"
    # t = text(sql, bindparams=[bindparam('alarm_time_param', type_=DateTime, required=True)])
    # db.execute(t, {"alarm_time_param": date_param})

    # 参数bindparam可以使用type_来指定参数的类型, 也可以使用 initial 值来指定参数类型
    #     bindparam('alarm_time_param', type_=DateTime) #直接指定参数类型
    #     bindparam('alarm_time_param', DateTime()) #使用初始值指定参数类型

    # 3.如要转换查询的结果中的数据类型, 可以通过text()
    # 的参数typemap参数指定.这点比mybatis还灵活,
    # t = text("SELECT id, name FROM users",
    #          typemap={
    #              'id': Integer,
    #              'name': Unicode
    #          }
    #          )

    # result = db.session.execute('select * from alist limit 5;')
    # print(result.fetchall())
    # print(result.fetchmany(5))
    # print(result.fetchone())
    # print(result.scalar())              # 返回数据条数
    # print(result.returns_rows)          # 判断是否有数据
    # print(result.rowcount)              # 更新/删除影响的行数

    #try:
    #   添加/删除/修改
    #   result.commit
    #ecept:
    #   result.rollback() 回滚数据

    # # sql 语句添加 也需要commit
    # tex = db.session.execute('insert into alist(name) values("bbbbbbbbb")')
    # db.session.commit()

flask-sqlchemy 连表增加方式

　　 # 表结构:
    # User       relationship表 字段名 article
    # Article    外键表 外键名 user_id

    # 通过relationship 添加数据 方式一
    # a = Article(title='bbbbbb', content='1dadasdad3')
    # u = User(email='4444444', password='123123123',article=[a])
    # db.session.add(u)
    # db.session.commit()

    # 通过relationship 添加数据 方式二
    # a = Article(title='bbbbbb', content='1dadasdad3')
    # u = User(email='4444444', password='123123123')
    # u.article.append(a)
    # db.session.add(u)
    # db.session.commit()

    # 上面方式的变种
    # u={
    #     'email':'eeeeee',
    #     'password':'qwdqdwqdqd'
    # }
    # a={
    #     'title':'qqq',
    #     'content':'qqq'
    # }
    # db.session.add(User(**u,article=[Article(**a)]))
    # db.session.commit()

flask-sqlchemy 删除操作

    # # 方式一
    # # db.session.query(传入表明),filter(筛选条件删除).delete()
    # db.session.query(User).filter(User.id == 7).delete()
    # db.session.commit()

    # # 方式二
    # u = User.query.filter(User.id==11).first()
    # db.session.delete(u)
    # db.session.commit()

flask-sqlchemy 修改操作

# 方式一:
    # u = User.query.filter(User.id == 10).first()
    # u.email = 'llllllllll'
    # u.password = 'qweqweqwe'
    # db.session.commit()

    aaa = {
        'email':'8989898989898',
        'password':'9999999999'
    }
    # 方式二:
    # u = User.query.filter(User.id == 10).update(aaa)
    # db.session.commit()

    # 方式三
    db.session.query(User).filter(User.id==10).update(aaa)
    db.session.commit()

简化开发过程

1.用Navicat Premium 模型设计数据，并将模型转为sql执行文件，并生成数据库

2.根据sqlacodegen 逆向模块 反向生成sqlalchemy 的ClassBean文件

UserBean文件

# coding: utf-8
from sqlalchemy import Column, DateTime, String
from datetime import datetime
from sqlalchemy.dialects.mysql import INTEGER, SMALLINT
from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()
metadata = Base.metadata


class User(Base):
    __tablename__ = 'user'

    id = Column(INTEGER(11), primary_key=True)
    username = Column(String(20), nullable=False)
    age = Column(INTEGER(11), nullable=False)
    sex = Column(SMALLINT(6))
    create_time = Column(DateTime, default=datetime.now())

# 先设计数据库，然后根据以下方式 输入当前的UserBean
# sqlacodegen --tables student(表名) --outfile ./UserBean.py(输出的文件名) mysql+pymysql://数据库用户名:密码@地址/数据库?charset=utf8
# sqlacodegen --tables user --outfile ./UserBean.py mysql+pymysql://root:123456@192.168.0.118:3306/test?charset=utf8"

3.在UserServer 文件中创建/绑定 数据表和Object之间的关系，然后就可以操作数据数据了。

UserServer 文件

from sqlalchemy import create_engine
from sqlalchemy.orm import Session
from sqlalchemy.ext.declarative import declarative_base
from UserBean import User

engine = create_engine("mysql+pymysql://root:123456@192.168.0.118:3306/test?charset=utf8")

# Base.metadata.drop_all(bind=engine)
Base = declarative_base()
Base.metadata.create_all(bind=engine)

s = Session(bind=engine)

# 新增操作
# try:
#     for i in range(1, 10):
#         user = User(username="apy_%s" % i, age=i, sex=0)
#         s.add(user)
#         s.commit()
# except Exception as e:
#     s.rollback()

# 查询操作
UserList = s.query(User)[0:5]
for i in UserList:
    print(i.username)