1. ORM

ORM，对象关系映射，对象和关系的映射，使用面向对象的方式来操作数据库。

关系模型和python对象之间的映射
table => class , 表映射为类
row => object ,行映射为实例
column => property ,字段映射为属性

2. SQLALchemy

SQLALchemy是一个ORM框架，SQLALchemy内部使用了连接池。安装: $ pip install sqlalchemy

**官方文档：**

查看版本：

import sqlalchemyprint(sqlalchemy.__version__)

2.1 创建连接

数据库连接的事情，交给引擎。

import sqlalchemy# https://docs.sqlalchemy.org/en/13/orm/tutorial.html,如何使用sqlalchemyfrom sqlalchemy import create_engineprint(sqlalchemy.__version__)IP = '127.0.0.1'USERNAME = 'root'PASSWORD = 'root'DB_NAME = 'test'PORT = 3306# mysql+pymysql://
   
    :
    
     @
     
      /
      
       [?
       
        ]# https://docs.sqlalchemy.org/en/13/dialects/mysql.html#module-sqlalchemy.dialects.mysql.pymysqlconn = "mysql+pymysql://{}:{}@{}:{}/{}".format(USERNAME, PASSWORD, IP, PORT, DB_NAME)# pymysql是驱动engine = create_engine(conn, echo=True)# echo=True 引擎是否打印执行的语句print(engine)# Engine(mysql+pymysql://root:***@127.0.0.1:3306/test)# lazy connecting: 懒连接，创建引擎不会马上连接数据库，直到让数据库执行任务时才连接

2.2 Declare a Mapping创建映射

2.2.1 创建基类

from sqlalchemy.ext.declarative import declarative_base# 创建基类，便于实体类继承。SQLALchemy大量使用了元编程Base = declartive_base()

2.2.2 创建实体类

from sqlalchemy import Column, Integer, Stringclass Student(Base):    __tablename__ = 'students'    # 字段名和属性名一致时，字段名可以省略不写    id = Column(Integer, primary_key=True)  # 定义字段类型和属性    name = Column(String(40), nullable=False)    age = Column(Integer, nullable=False)    # age = Column('age', Integer, nullable=False)    def __repr__(self):        # return "<{} id={}, name={}, age={}>".format(__class__.__name__, self.id, self.name, self.age)        # 推荐使用C风格的写法        return "
   
    " % (self.id, self.name, self.age)print(Student.__dict__)print(repr(Student.__table__)

实例化：

student = Student(name='James', age=35)print(student.name, student.age, student.id)

创建表：

可以使用SQLALchemy来创建表、删除表

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

注意：生产环境很少这样创建表，都是系统上线的时候由脚本生成。生产环境很少删除表，宁可废弃都不删除。

创建会话session：

在一个会话中操作数据库，会话建立在连接上，连接被引擎管理。当第一次使用数据库时，从引擎维护的连接池中获取一个连接使用。

from sqlalchemy.orm import sessionmaker# 创建sessionSession = sessionmaker(bind=engine)  # 工厂方法返回类session = Session()  # 实例化# 依然在第一次使用时连接数据库

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

CRUD操作：

增

add():增加一个对象

add_all: 可以增加多个对象，元素是可迭代对象

import sqlalchemy# https://docs.sqlalchemy.org/en/13/orm/tutorial.html,如何使用sqlalchemyfrom sqlalchemy import create_enginefrom sqlalchemy.ext.declarative import declarative_basefrom sqlalchemy import Column, Integer, Stringfrom sqlalchemy.orm import sessionmaker print(sqlalchemy.__version__)IP = '127.0.0.1'USERNAME = 'root'PASSWORD = 'root'DB_NAME = 'test'PORT = 3306# mysql+pymysql://
   
    :
    
     @
     
      /
      
       [?
       
        ]# https://docs.sqlalchemy.org/en/13/dialects/mysql.html#module-sqlalchemy.dialects.mysql.pymysqlconn = "mysql+pymysql://{}:{}@{}:{}/{}".format(USERNAME, PASSWORD, IP, PORT, DB_NAME)# pymysql是驱动engine = create_engine(conn, echo=True)# echo=True 引擎是否打印执行的语句print(engine)# Engine(mysql+pymysql://root:***@127.0.0.1:3306/test)# lazy connecting: 懒连接，创建引擎不会马上连接数据库，直到让数据库执行任务时才连接Base = declarative_base()  # 基类class Student(Base):    __tablename__ = 'students'    # 字段名和属性名一致时，字段名可以省略不写    id = Column(Integer, primary_key=True)  # 定义字段类型和属性    name = Column(String(40), nullable=False)    age = Column(Integer, nullable=False)    # age = Column('age', Integer, nullable=False)    def __repr__(self):        # return "<{} id={}, name={}, age={}>".format(__class__.__name__, self.id, self.name, self.age)        # 推荐使用C风格的写法        return "
        
         " % (self.id, self.name, self.age)print(Student.__dict__)# print(Student.__table__) # 没有出自己想要的结果（默认调用str方法），可以调用一下repr方法print(repr(Student.__table__))Base.metadata.create_all(engine) # 创建继承自Base的所有表# Base.metadata.drop_all(engine)student = Student(name='James', age=35)print(student.name, student.age, student.id)session = sessionmaker(bind=engine)() # 线程不安全，不适用多线程student = Student(id=1, name='James', age=28)session.add(student)session.commit()try: student.name = 'Klay' session.add_all([ Student(id=2, name='Curry', age=31), Student(id=3, name='KD', age=32), Student(id=4, name='Green', age=29) ]) session.commit() print('~~~~~~~~~~~~~~~')except Exception as e: session.rollback() print(e) print('+++++++++++++++++')

简单查询：

students = session.query(Student)print(students)  # 无内容，惰性的for s in students:    print(s)print('-----------------')"""\
   
    
     
      
       """student = session.query(Student).get(2)  # 通过主键查询print(student)#

query方法将实体类传入，返回类的对象（可迭代对象），这时候并不查询，迭代它就执行SQL来查询数据库，封装数据到指定的类的实例。get方法使用主键查询，返回一条传入类的一个实例。

改：

student = session.query(Student).get(3)  # 必须先查出来，然后修改，最后在提交print(student)# 
   
    student.name = 'Kevin'student.age = 32print(student)# 
    
     session.add(student)session.commit()

状态 ***

每一个实体，都有一个状态属性_sa_instance_state,其类型是sqlalchemy.orm.state.InstanceState,可以使用sqlalchemy.inspect(entity)函数查看状态。常见的状态值有transient、peding、persistent、deleted、detached。

新建一个实体，状态就是transient临时的，一旦add()后从transient变成pending状态。成功提交后从pending变成persistent状态，成功查询返回的实体对象也是persistent状态，persistent状态的实体，修改后还是persistent状态，persistent状态的实体，删除后，flush后没有commit，就变成deleted状态，成功提交，就变成detached状态，提交失败，还原到persistent状态。flush方法，主要是把改变应用到数据库中去。

import sqlalchemy# https://docs.sqlalchemy.org/en/13/orm/tutorial.html,如何使用sqlalchemyfrom sqlalchemy import create_enginefrom sqlalchemy.ext.declarative import declarative_basefrom sqlalchemy import Column, Integer, Stringfrom sqlalchemy.orm import sessionmakerfrom sqlalchemy.orm.state import InstanceStateprint(sqlalchemy.__version__)IP = '127.0.0.1'USERNAME = 'root'PASSWORD = 'root'DB_NAME = 'test'PORT = 3306# mysql+pymysql://
   
    :
    
     @
     
      /
      
       [?
       
        ]# https://docs.sqlalchemy.org/en/13/dialects/mysql.html#module-sqlalchemy.dialects.mysql.pymysqlconn = "mysql+pymysql://{}:{}@{}:{}/{}".format(USERNAME, PASSWORD, IP, PORT, DB_NAME)# pymysql是驱动engine = create_engine(conn, echo=True)# echo=True 引擎是否打印执行的语句print(engine)# Engine(mysql+pymysql://root:***@127.0.0.1:3306/test)# lazy connecting: 懒连接，创建引擎不会马上连接数据库，直到让数据库执行任务时才连接Base = declarative_base()  # 基类class Student(Base):    __tablename__ = 'students'    # 字段名和属性名一致时，字段名可以省略不写    id = Column(Integer, primary_key=True)  # 定义字段类型和属性    name = Column(String(40), nullable=False)    age = Column(Integer, nullable=False)    # age = Column('age', Integer, nullable=False)    def __repr__(self):        # return "<{} id={}, name={}, age={}>".format(__class__.__name__, self.id, self.name, self.age)        # 推荐使用C风格的写法        return "
        
         " % (self.id, self.name, self.age)session = sessionmaker(bind=engine)()def get_state(instance, i): state: InstanceState = sqlalchemy.inspect(instance) output = "{} :{} {} \n"\ "attached={}, transient={}, pending={}\n"\ "persistent={}, deleted={}, detached={}\n".format(i, state.key, state.session_id, state._attached, state.transient, state.pending, state.persistent, state.deleted, state.detached) print(output, end="~~~~~~~~~~~~~~~\n")student = session.query(Student).get(2)get_state(student, 1) # persistenttry: student = Student(id=5, name='AD', age=28) get_state(student, 2) # transient student = Student(id=6, name='Durant', age=32) get_state(student, 3) # transient session.add(student) get_state(student, 4) # pending session.commit() get_state(student, 5) # persistent except Exception as e: session.rollback() print(e, '~~~~~~~~~~~~')student = session.query(Student).get(6)get_state(student, 6) # persistenttry: session.delete(student) get_state(student, 7) # persistent(未flush) session.flush() get_state(student, 8) # deleted session.commit() get_state(student, 9) # detached(分离)except Exception as e: session.rollback() print('++++++++++++++++') print(e)

复杂查询

from sqlalchemy import create_engine, Column, String, Integer, Enum, Date, ForeignKeyfrom sqlalchemy.ext.declarative import declarative_basefrom sqlalchemy.orm import sessionmaker, relationship  # , scoped_session# from sqlalchemy import and_, or_, not_from sqlalchemy import funcimport enumBase = declarative_base()  # 基类IP = '127.0.0.1'USERNAME = 'root'PASSWORD = 'root'DB_NAME = 'test'PORT = 3306# mysql+pymysql://
   
    :
    
     @
     
      /
      
       [?
       
        ]# https://docs.sqlalchemy.org/en/13/dialects/mysql.html#module-sqlalchemy.dialects.mysql.pymysqlconn = "mysql+pymysql://{}:{}@{}:{}/{}".format(USERNAME, PASSWORD, IP, PORT, DB_NAME)# pymysql 是驱动engine = create_engine(conn, echo=True)session = sessionmaker(bind=engine)()class MyEnum(enum.Enum):  # 枚举一般不用，建议直接写0，1    F = 'F'    M = 'M'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)    departments = relationship('DeptEmp')    def __repr__(self):        return "<{} emp_no={}, name={}, gender={}, dept={}>".format(__class__.__name__, self.emp_no,                                                                    "{} {}".format(self.first_name, self.last_name),                                                                    self.gender.value, self.departments)class Department(Base):    __tablename__ = 'departments'    dept_no = Column(Integer, primary_key=True)    dept_name = Column(String(40), unique=True)    def __repr__(self):        return "<{} dept_no={}, dept_name={}>".format(__class__.__name__, self.dept_no, self.dept_name)class DeptEmp(Base):    __tablename__ = 'dept_emp'    emp_no = Column(Integer, ForeignKey('employees.emp_no', ondelete='CASCADE'), primary_key=True)    dept_no = Column(Integer, 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 "<{} emp_no={}, dept_no={}>".format(__class__.__name__, self.emp_no, self.dept_no)def show(ems):    for x in ems:        print(x)    print('\n\n')# eps = session.query(Employee).filter(Employee.emp_no > 10015)  # filter相当于where# # query可以理解为select * from# show(eps)# and or not# eps = session.query(Employee).filter(Employee.emp_no > 10015, Employee.emp_no < 10019)# show(eps)# eps = session.query(Employee).filter(Employee.emp_no > 10015).filter(Employee.emp_no < 10019)# show(eps)# eps = session.query(Employee).filter((Employee.emp_no > 10015) & (Employee.emp_no < 10019))# show(eps)# eps = session.query(Employee).filter(and_(Employee.emp_no > 10015, Employee.emp_no < 10019))# show(eps)## es = session.query(Employee).filter(or_(Employee.emp_no == 10015, Employee.emp_no == 10018))# show(es)## # es = session.query(Employee).filter((Employee.emp_no == 10015) | (Employee.emp_no == 10018))# # show(es)## ss = session.query(Employee).filter(not_(Employee.emp_no < 10019))# show(ss)# ss = session.query(Employee).filter(~(Employee.emp_no < 10019))# show(ss)# # in, 注意调用的是in_函数# emp_list = (10010, 10019, 10021)## esm = session.query(Employee).filter(Employee.emp_no.in_(emp_list))# show(esm)## # # not in# # ems = session.query(Employee).filter(~(Employee.emp_no.in_(emp_list)))# # show(ems)## # like# ms = session.query(Employee).filter(Employee.first_name.like('P%'))# show(ms)### # # not like,ilike默认不区分大小写# # mse = session.query(Employee).filter(Employee.last_name.notlike('M%'))# # show(mse)## sm = session.query(Employee).filter(Employee.emp_no > 10017).order_by(Employee.emp_no)# show(sm)## sme = session.query(Employee).filter(Employee.emp_no > 10017).\#     order_by(Employee.gender).order_by(Employee.emp_no.desc())# # 注意Employees.gender不是按照升序排列的，而是类变量的定义顺序，暂时还是有问题，明天继续探索？# show(sme)# 分页sem = session.query(Employee).filter(Employee.emp_no > 10015).limit(2).offset(2)show(sem)# 消费者方法,消费者方法调用后，Query对象就转换成了一个容器ps = session.query(Employee)# print(ps.count())  # 聚合函数count(*)的查询，用了子查询，效率不高print(len(list(ps)))# for i in ps:#     print(i)print(ps.all())  # 结果是列表，列表中是21个对象，查不到返回空列表print(ps.first())  # 返回行首，查不到返回None# print(ps.one())  # 有且只能有一行，如果查询结果是多行则抛异常# 删除 delete# session.query(Employee).filter(Employee.emp_no > 10019).delete()# session.commit()  # 提交后就真的删除了# # 聚合、分组# query = session.query(func.count(Employee.emp_no))# print(query.all())  # [(21,)]，结果是列表# print(query.first())  # (21,)，结果是元组# print(query.one())  # (21,),结果是元组# print(query.scalar())  # 21，取one的第一个元素query = session.query(Employee.gender, func.count(Employee.emp_no)).group_by(Employee.gender).all()print(query)# [(
        
         , 13), (
         
          , 8)]for g, y in query: print(g) print(g.value, y)print("~~~~~~~~~~~~~~~")# 关联查询# 查询10010员工的所在部门编号及员工信息result = session.query(Employee, DeptEmp).filter(Employee.emp_no == DeptEmp.emp_no, Employee.emp_no == 10010).all()# 隐式连接等价于"""\SELECT *from employees, dept_empwhere employees.emp_no = dept_emp.emp_no and employees.emp_no = 10018;"""show(result) # 结果是两个对象# 使用join# results = session.query(Employee).join(DeptEmp).filter(Employee.emp_no == 10010).all()# show(results)# results = session.query(Employee).join(DeptEmp, Employee.emp_no == DeptEmp.emp_no).\# filter(Employee.emp_no == 10010).all()## show(results)results = session.query(Employee).join(DeptEmp, (Employee.emp_no == DeptEmp.emp_no) & (Employee.emp_no == 10010))show(results.all()) # 注意返回的结果是一个对象# 
          
           ,# 
           
            ]>for j in results: print(j) print(j.emp_no) # 10010 print(j.departments) # [
            
             , 
             
              ]

总结：

在开发中，一般都会采用ORM框架，这样就可以使用对象操作表了。定义表映射的类，使用Column的描述器定义类属性，使用ForeignKey来定义外键约束。如果在一个对象中，想查看其它表对应的对象的内容，就要使用relationship来定义关系。

转载地址：http://gpfvi.baihongyu.com/

你可能感兴趣的文章