Python ORM框架之sqlalchemy

一、概念

什么是ORM框架？
ORM：Object Relation Mapping，最初主要描述的是程序中的Object对象和关系型数据库中Rlation关系(表)之间的映射关系，目前来说也是描述程序中对象和数据库中数据记录之间的映射关系的统称，是一种进行程序和数据库之间数据持久化的一种编程思想。
什么是sqlalchemy？
sqlalchemy是一个python语言实现的的针对关系型数据库的orm库。可用于连接大多数常见的数据库，比如MySQL、SQLite、mongodb，redis等。三方库，pip install sqlalchemy

二、具体使用方法

1. 数据库—表操作

创建数据库，创建表格，表格字段获取：

# -*- coding:utf-8 -*-

import os, random
from datetime import date
import sqlalchemy
from sqlalchemy.engine import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column
from sqlalchemy.types import String, Date, Float, Integer
from sqlalchemy import event, DDL
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

current_path = os.path.join(os.path.dirname(os.path.abspath(__file__)))
db_path = os.path.join(current_path, 'dbs', 'student.db')
print(db_path)

Base = declarative_base()

#字段说明： https://blog.csdn.net/gymaisyl/article/details/86508244
class User(Base):
    __tablename__ = 'user'

    id = Column(Integer, primary_key=True, autoincrement=True)#默认从1开始
    name = Column(String(32), nullable=False)
    age = Column(Integer, nullable=False)
    birthday = Column(Date)
    score = Column(Float)

    def __repr__(self):
        return self.name

def create_engin():
    ''' 创建数据库引擎 获取数据库中的表名列表，若表不存在，则新建 pymysql mysql+pymysql://<username>:<password>@<host>/<dbname>[?<options>] engine = create_engine("mysql+pymysql://root:123456@localhost:3306/db4?charset=utf8", max_overflow=0, # 超过连接池大小外最多创建的连接 pool_size=5, # 连接池大小 pool_timeout=30, # 池中没有线程最多等待的时间，否则报错 pool_recycle=-1 # 多久之后对线程池中的线程进行一次连接的回收（重置） echo = True # echo参数为True时，会显示每条执行的SQL语句，可以关闭 ", max_overflow = 5) sqlite3， sqlite3 python标准库 engine = create_engine('sqlite:///%s'%db_path) '''

    engine = create_engine('sqlite:///%s'%db_path)
    db_insp = sqlalchemy.inspect(engine)
    table_name_list = db_insp.get_table_names()
    print(table_name_list)
    print('*'*200)
    if User.__tablename__ not in table_name_list:
        Base.metadata.create_all(engine) #创建表
        #Base.metadata.drop_all(engine) #删除现有的表，谨慎决定是否需要这样操作
        ##设置id自增开始位置 1001
        event.listen(
            User.__table__,
            'after_create',
            DDL("ALTER TABLE %(table) AUTO_INCREMENT = 1001;")
        )
        print("create table successful!")

    else:
        print('table is exist!')
    return engine

def structure(engine):
    ''' 该函数目的是查询数据库，表信息 '''
    db_insp = sqlalchemy.inspect(engine) # return sqlalchemy.engine.reflection.Inspector
    print(db_insp)
    print(db_insp.get_table_names())  # 获取所有表格名字，return list

    md = sqlalchemy.MetaData() # return MetaData
    table = sqlalchemy.Table('user', md, autoload=True, autoload_with=engine)
    print(table.c)
    print(table.c.id.type)

2. 数据库—表内容操作

表格内容操作，无非就是基于session的增、删、改、查

# -*- coding:utf-8 -*-

import os, random
from datetime import date
import sqlalchemy
from sqlalchemy.engine import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column
from sqlalchemy.types import String, Date, Float, Integer
from sqlalchemy import event, DDL
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

def create_session(engine):
    ''' 创建 session '''
    Session = sessionmaker(bind=engine)
    session = Session()
    return session

def insert_data(session):
    ''' 插入数据 '''
    try:
        ##单条数据插入，id为自增类型，不用指定
        user_obj = User()
        user_obj.name = 'zhangsan'
        user_obj.age = 30
        user_obj.birthday = date(year=1992, month=3, day=10)
        user_obj.score = 95
        session.add(user_obj)

        ##批量插入数据
        names = ['jack'+str(i) for i in range(1,100)]
        print(names)

        ages = []
        for i in range(100):
            ages.append(random.randint(18, 70))
        print(ages)

        birthdays = []
        for i in range(100):
            birthdays.append(date(year=random.randint(1980, 2020), month=random.randint(1,12), day=random.randint(1, 29)))
        print(birthdays)

        scores = []
        for i in range(100):
            scores.append(round(random.random()*100, 1))
        print(scores)

        user_list = []
        for i in range(len(names)):
            user = User()
            user.name = names[i]
            user.age = ages[i]
            user.birthday = birthdays[i]
            user.score = scores[i]

            user_list.append(user)

        print(user_list)
        session.add_all(user_list)
        session.commit()
        print("insert data successful!")
    except Exception as e:
        print(e)
    finally:
        session.close()

def query_data(session):
    ''' session.query(User.age).filter(User.name=='Jack'): 说明：query()中指定返回字段；filter()中指定筛选条件 sqlalchemy.orm.query.Query.func(): func() == all() 返回所有数据 func() == first() 返回第一条数据 func() == scalar()/one() 返回一条数据，若多行，则报异常 func() == one_or_none() 返回一条数据，若多行，则返回none func.count ：统计行数 func.avg：求平均 func.max：求最大 func.min：求最小 func.sum：求和 func.limit(number) 限制返回的记录数量 '''
    print('*'*80+'query'+'*'*80)
    query = session.query(User.id, User.name, User.score) #return <class 'sqlalchemy.orm.query.Query'>
    print(query.all()) #return [(),(),()]

    query_24 = query.filter(User.age == 24)
    print(query_24.first())
    print(query_24.all())
    
    ### one()只返回一条数据，若返回多条数据，则报异常，与scalar()一样
    user_jack = session.query(User.age).filter(User.name=='Jack').one()
    user_jack = session.query(User.age).filter(User.name=='Jack').scalar()
    print(user_jack)#return (24,) #查询名字为jack的人的年龄

    ### one()只返回一条数据，若返回多条数据，则报返回None
    user_jack = session.query(User.age).filter(User.name=='Jack').one_or_none()
    print(user_jack)#return (24,) #查询名字为jack的人的年龄

def update_data(session):
    ''' 该函数用于更改数据 '''
    jack_obj = session.query(User).filter(User.name == "Jack").first()
    print(type(jack_obj)) #return <class '__main__.User'>
    jack_obj.score = 100
    session.commit()
    session.close()
    print("update data successful!")

def drop_data(session):
    ''' 该函数用删除数据 '''
    jack99_obj = session.query(User).filter(User.name == "jack99").first()
    print(jack99_obj)
    session.delete(jack99_obj)
    session.commit()
    session.close()
    print('delete data successful!')

if __name__ == '__main__':
    engine = create_engin()
    structure(engine)
    session = create_session(engine)
    #insert_data(session)
    query_data(session)
    #update_data(session)
    #drop_data(session)