MySQL—ORM框架，sqlalchemy模块

Posted 2021-01-13 阿橙

武老师博客：ORM框架介绍

 

import os
#1.当一类函数公用同样参数时候，可以转变成类运行 - 分类
#2.面向对象： 数据和逻辑组合在一起了
#3. 一类事物共同用有的属性和行为（方法）


#因此 表其实可以写成一个类
#双下方法item 和 call  必须要背会：

class Userinfo:
    def __init__(self,id,name):
        self.id = id
        self.name = name
    def show(self):
        print(\'in the show\')

    def __call__(self, *args, **kwargs):
        print(\'你在调用call双下方法哦\')

    def __getitem__(self, item):
        return self.__dict__[item]

    def __setitem__(self,key,value):
        self.__dict__[key] = value

    def __delitem__(self, key):
        del self.__dict__[key]

user1 = Userinfo(1,\'gkx\')  #类似表第一行

user1()
user1[\'age\'] = 22
print(user1.__dict__)
print(user1[\'id\'])
# del user1[\'age\']  #调用 __delitem__ 方法
del user1.age  #原生方法
print(user1.__dict__)

面向对象复习

#ORM : object relational mapper
#1.ORM框架： ORM-->关系对象映射 ： SQLAlchemy：code first,默认没有db first，想有，要装第三方工具
    #作用
        #1.提供简单的规则
        #2.自动转换成sql语句

    #两类
        # DB first / code first 所有的ORM都是创建一个类，让类去对应表
\'\'\'
  DB first:  帮你从db中把已创建好的表，连上db后，自动把表转化为代码中的一个个类
Code first： 先有类和db，然后在数据库创建表  

- DB first: 手动创建数据库以及表          -> ORM框架 -> 自动生成类
- code first： 手动创建类、和数据库        -> ORM框架 -> 以及表  SQLAlchemy 
\'\'\'


#20181011 ORM框架 图在有道词典
#SQLAlchemy 不进行连接，要用pymysql等进行连接
\'\'\'
SQLAlchemy是Python编程语言下的一款ORM框架，该框架建立在数据库API之上，使用关系对象映射进行数据库操作，
            简言之便是：将对象转换成SQL，然后使用数据API执行SQL并获取执行结果。
SQLAlchemy本身无法操作数据库，其必须以来pymsql等第三方插件，Dialect用于和数据API进行交流，根据配置文件的不同调用不同的数据库API，从而实现对数据库的操作，如：

MySQL-Python
    mysql+mysqldb://<user>:<password>@<host>[:<port>]/<dbname>
   
pymysql
    mysql+pymysql://<username>:<password>@<host>/<dbname>[?<options>]
   
MySQL-Connector
    mysql+mysqlconnector://<user>:<password>@<host>[:<port>]/<dbname>
   
cx_Oracle
    oracle+cx_oracle://user:pass@host:port/dbname[?key=value&key=value...]
   
更多详见：http://docs.sqlalchemy.org/en/latest/dialects/index.html
\'\'\'

ORM框架初识

sqlalchemy是codefirst
　　DB first: 帮你从db中把已创建好的表，连上db后，自动把表转化为代码中的一个个类
　　Code first： 先有类和db，然后在数据库创建表

　　- DB first: 手动创建数据库以及表 -> ORM框架 -> 自动生成类
　　- code first： 手动创建类、和数据库 -> ORM框架 -> 以及表 SQLAlchemy

 

 

from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String, ForeignKey, UniqueConstraint, Index
from sqlalchemy.orm import sessionmaker, relationship
from sqlalchemy import create_engine

Base = declarative_base()

class UserType(Base):
    __tablename__ =  \'usertype\'
    id = Column(Integer,primary_key=True,autoincrement=True)
    title = Column(String(32),nullable=True,index=True)

class Users(Base):
    __tablename__ =  \'users\'
    id = Column(Integer,primary_key=True,autoincrement=True)
    name = Column(String(32),nullable=True,index=True)
    email = Column(String(16),unique=True)
    user_type_id = Column(Integer,ForeignKey("usertype.id"))

    # __table_args__ = (
    #     UniqueConstraint(\'id\',\'name\',name=\'uix_id_name\'),
    #     Index(\'ix_n_ex\',\'name\',\'email\')
    # )

def create_db():
    engine = create_engine("mysql+pymysql://root:gkx321@127.0.0.1:3306/ORMdb?charset=utf8", max_overflow=5)
    Base.metadata.create_all(engine)
    # Base.metadata.drop_all(engine)  #删除类对应的所有表

def drop_db():
    engine = create_engine("mysql+pymysql://root:gkx321@127.0.0.1:3306/ORMdb?charset=utf8", max_overflow=5)
    Base.metadata.drop_all(engine)  #删除类对应的所有表

engine = create_engine("mysql+pymysql://root:gkx321@127.0.0.1:3306/ORMdb?charset=utf8", max_overflow=5)
Session = sessionmaker(bind=engine)
session = Session() #类似pymysql中conn和cursor的综合

#****增加****************
# obj1 = UserType(title=\'普通用户\')
# session.add(obj1)
# objs = [
#     UserType(title=\'白银用户\'),
#     UserType(title=\'黄金用户\')
# ]
# session.add_all(objs)

#****查询****************
\'\'\'
usertype_lst = session.query(UserType).all()
print(session.query(UserType)) #----->> SELECT usertype.id AS usertype_id, usertype.title AS usertype_title FROM usertype
print(usertype_lst) #一个列表，包含了usertype类中的每一个对象，而这每一个对象就对应usertype表的一行数据
for i in usertype_lst:
    print(i.id,i.title)#直接打印i会是一个对象的内存地址，应该这么打印才有值
# ···usertype_lst = session.query(UserType.id).all() #如果query里面不是一个类，而是一个属性，那么 usertype_lst是有值的，可以直接打印
\'\'\'
# usertype_lst = session.query(UserType.id,UserType.title).filter(UserType.id > 2) #或者 UserType.id ==2
# for i in usertype_lst:
#     print(i.title)

#****删除****************
# session.query(UserType.id,UserType.title).filter(UserType.id > 2).delete()

#****修改****************
# session.query(UserType.id,UserType.title).filter(UserType.id > 0).update({\'title\':\'黑金\'})
# session.query(UserType.id,UserType.title).filter(UserType.id > 0).update({UserType.title:UserType.title + \'x\'},synchronize_session=False)
# session.query(UserType.id,UserType.title).filter(UserType.id > 0).update({\'title\':UserType.title + 1},synchronize_session="evaluate")









session.commit()
session.close()

用sqlalchemy创建类，及增删该查

query: 类似select
filter：类似 where
【.】 : 类似sql里的 空格 用来分割命令的
all ： fetchone，fetchall，如果不加all，那么打印ret就只是返回一句sql语句
#如果query里是一个类，all的取值是一个个对象，如果query里是一个 类.属性 那么ret可以直接打印出来

 

子查询 select id,(select name from tb2 where id = xx) from tb2
#sqlalchemy 支持 分组，排序，连表，通配符，子查询，limit，union，where
#如果语句很复杂，还支持 原生sql

#子查询 select id,(select name from tb2 where id = xx) from tb2
#sqlalchemy 支持 分组，排序，连表，通配符，子查询，limit，union，where
#如果语句很复杂，还支持 原生sql

from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String, ForeignKey, UniqueConstraint, Index
from sqlalchemy.orm import sessionmaker, relationship
from sqlalchemy import create_engine

Base = declarative_base()

engine = create_engine("mysql+pymysql://root:gkx321@127.0.0.1:3306/ORMdb?charset=utf8", max_overflow=5)
Session = sessionmaker(bind=engine)
session = Session() #类似pymysql中conn和cursor的综合

\'\'\'
#　条件
ret = session.query(Users).filter_by(name=\'alex\').all()  #filter_by内部调用的其实是filter，只是这里用参数表示， filter Users.name=\'alex\' 是表达式
ret = session.query(Users).filter(Users.id > 1, Users.name == \'eric\').all()  #默认是and，除非声明是or 见下方导入 or_
ret = session.query(Users).filter(Users.id.between(1, 3), Users.name == \'eric\').all()
ret = session.query(Users).filter(Users.id.in_([1,3,4])).all()               # 人家in_就是规定要这么写啊
ret = session.query(Users).filter(~Users.id.in_([1,3,4])).all()              # ~ 表示非 not in
ret = session.query(Users).filter(Users.id.in_(session.query(Users.id).filter_by(name=\'eric\'))).all()

from sqlalchemy import and_, or_
ret = session.query(Users).filter(and_(Users.id > 3, Users.name == \'eric\')).all()  #因为默认是and，此处不声明 and_也可以的
ret = session.query(Users).filter(or_(Users.id < 2, Users.name == \'eric\')).all()
ret = session.query(Users).filter(
    or_(
        Users.id < 2,
        and_(Users.name == \'eric\', Users.id > 3),
        Users.extra != ""
    )).all()


# 通配符
ret = session.query(Users).filter(Users.name.like(\'e%\')).all()    # 不要忘了通配符还有一个是 _ 表示一位
ret = session.query(Users).filter(~Users.name.like(\'e%\')).all()   # ~ 非的意思啦

# 限制
ret = session.query(Users)[1:2]   # limit 1,2

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

# 分组
from sqlalchemy.sql import func  #使用函数要导入函数模块！

ret = session.query(Users).group_by(Users.extra).all()
ret = session.query(
    func.max(Users.id),
    func.sum(Users.id),
    func.min(Users.id)).group_by(Users.name).all()

ret = session.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 = session.query(Users, Favor).filter(Users.id == Favor.nid).all()  # select * from Users,Favor where Users.id = Favor.nid #相当于inner join

ret = session.query(Person).join(Favor).all()   #如果两个表有设置了foreign key ，那么不用写条件，自动关联了 inner join
ret = session.query(Person).join(Favor,isouter=True).all()   #left join, right join呢？ 表换一下位置就好了。。。

ret = session.query(Person).join(Favor, isouter=True).all()


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

q1 = session.query(Users.name).filter(Users.id > 2)
q2 = session.query(Favor.caption).filter(Favor.nid < 2)
ret = q1.union_all(q2).all()
\'\'\'

#子查询的正确写法 ： .subquery()   .as_acalar()
# 1.
# select * from b where id in (select id from tb2)

# 2 select * from (select * from tb) as B
# q1 = session.query(UserType).filter(UserType.id > 0).subquery()
# result = session.query(q1).all()
# print(result)

# 3
# select
#   id ,
#   (select * from users where users.user_type_id=usertype.id)
# from usertype;
#第一次循环 拿到 id1  然后判断id1与另一张表相等，拿到一个值
#第二次循环 拿到 id2  然后判断id2与另一张表相等，拿到一个值...

# session.query(UserType,session.query(Users).filter(Users.id == 1).subquery()) #错误写法
# session.query(UserType,Users) #上面这么写，其实还是类似笛卡儿积，达不到子查询的效果

# result = session.query(UserType.id,session.query(Users).as_scalar())
# print(result)

# result = session.query(UserType.id,session.query(Users).filter(Users.user_type_id==UserType.id).as_scalar())
# print(result)

session.commit()
session.close()

sqlalchemy—子查询

 

通过设置relationship可以减少连表带来的冗余代码，只要通过 obj.relationship_name 就可以获取表格整行数据

# 谁有foreign key relationship就写在哪里
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String, ForeignKey, UniqueConstraint, Index
from sqlalchemy.orm import sessionmaker, relationship
from sqlalchemy import create_engine

Base = declarative_base()

engine = create_engine("mysql+pymysql://root:gkx321@127.0.0.1:3306/ORMdb?charset=utf8", max_overflow=5)
Session = sessionmaker(bind=engine)
session = Session() #类似pymysql中conn和cursor的综合

\'\'\'
获取用户信息以及与其关联的用户名称： 用SQL的时候就是要先连表
\'\'\'
# 问题1. 获取用户信息以及与其关联的用户类型名称(FK,Relationship=>正向操作)
# user_list = session.query(Users,UserType).join(UserType,isouter=True)
# print(user_list)
# for row in user_list:
#     print(row[0].id,row[0].name,row[0].email,row[0].user_type_id,row[1].title)

# user_list = session.query(Users.name,UserType.title).join(UserType,isouter=True).all()
# for row in user_list:
#     print(row[0],row[1],row.name,row.title)




\'\'\'谁有foreign key relationship就写在哪里
relationalship~~~~
class Users(Base):
    __tablename__ = \'users\'
    id = Column(Integer, primary_key=True, autoincrement=True)
    name = Column(VARCHAR(32), nullable=True, index=True)
    email = Column(VARCHAR(16), unique=True)
    user_type_id = Column(Integer,ForeignKey("usertype.id"))

    user_type = relationship("UserType",backref=\'xxoo\') ***********创建关系！ 这个backref是给usertype用的，即父表用的

然后就可以使用以下方法来获取了，不用连表这么麻烦了\'\'\'
# user_list = session.query(Users)
# for row in user_list:
#     print(row.name,row.id,row.user_type.title)


# 问题2. 获取用户类型
# type_list = session.query(UserType) 传统方式
# for row in type_list:
#     print(row.id,row.title,session.query(Users).filter(Users.user_type_id == row.id).all())

# type_list = session.query(UserType)
# for row in type_list:
#     print(row.id,row.title,row.xxoo)

# 设置了backref后
# row.xxoo 就相当于 session.query(Users).filter(Users.user_type_id == row.id).all()

"""tb1
1   白金
2   黑金
obj.xx ==> [obj,obj...]   #这叫做反向操作，设置了 backref=\'xx\'后
"""

"""tb2
1   方少伟   1
2   成套     1
3   小白     2
# 正向
ut = relationship（backref=\'xx\'）
obj.ut ==> 相当于拿到 tb1 在id对应位置一整行数据  这叫做正向操作
"""

session.commit()
session.close()

ORM： relationship