Flask-SQLAlchemy-白红宇

Flask-SQLAlchemy

阅读量：2177 次

发布时间：2019-05-01

本文共 17043 字，大约阅读时间需要 56 分钟。

一. 介绍

SQLAlchemy是一个基于Python实现的ORM框架。该框架建立在 DB API之上，使用关系对象映射进行数据库操作，简言之便是：将类和对象转换成SQL，然后使用数据API执行SQL并获取执行结果。

pip3 install sqlalchemy

组成部分：

Engine，框架的引擎

Connection Pooling ，数据库连接池

Dialect，选择连接数据库的DB API种类

Schema/Types，架构和类型

SQL Exprression Language，SQL表达式语言

SQLAlchemy本身无法操作数据库，其必须以来pymsql等第三方插件，Dialect用于和数据API进行交流，根据配置文件的不同调用不同的数据库API，从而实现对数据库的操作，如：

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

二. 使用

1. 执行原生SQL语句

import timeimport threadingimport sqlalchemyfrom sqlalchemy import create_enginefrom sqlalchemy.engine.base import Engineengine = create_engine(    "mysql+pymysql://root:123@127.0.0.1:3306/t1?charset=utf8",    max_overflow=0,  # 超过连接池大小外最多创建的连接    pool_size=5,  # 连接池大小    pool_timeout=30,  # 池中没有线程最多等待的时间，否则报错    pool_recycle=-1  # 多久之后对线程池中的线程进行一次连接的回收（重置）)def task(arg):    conn = engine.raw_connection()    cursor = conn.cursor()    cursor.execute(        "select * from t1"    )    result = cursor.fetchall()    cursor.close()    conn.close()for i in range(20):    t = threading.Thread(target=task, args=(i,))    t.start()

#!/usr/bin/env python# -*- coding:utf-8 -*-import timeimport threadingimport sqlalchemyfrom sqlalchemy import create_enginefrom sqlalchemy.engine.base import Engineengine = create_engine("mysql+pymysql://root:123@127.0.0.1:3306/t1", max_overflow=0, pool_size=5)def task(arg):    conn = engine.contextual_connect()    with conn:        cur = conn.execute(            "select * from t1"        )        result = cur.fetchall()        print(result)for i in range(20):    t = threading.Thread(target=task, args=(i,))    t.start()

View Code

#!/usr/bin/env python# -*- coding:utf-8 -*-import timeimport threadingimport sqlalchemyfrom sqlalchemy import create_enginefrom sqlalchemy.engine.base import Enginefrom sqlalchemy.engine.result import ResultProxyengine = create_engine("mysql+pymysql://root:123@127.0.0.1:3306/t1", max_overflow=0, pool_size=5)def task(arg):    cur = engine.execute("select * from t1")    result = cur.fetchall()    cur.close()    print(result)for i in range(20):    t = threading.Thread(target=task, args=(i,))    t.start()

View Code

注意: 查看连接 show status like 'Threads%';

2. ORM

a. 创建数据库表

#!/usr/bin/env python# -*- coding:utf-8 -*-import datetimefrom sqlalchemy import create_enginefrom sqlalchemy.ext.declarative import declarative_basefrom sqlalchemy import Column, Integer, String, Text, ForeignKey, DateTime, UniqueConstraint, IndexBase = declarative_base()class Users(Base):    __tablename__ = 'users'    id = Column(Integer, primary_key=True)    name = Column(String(32), index=True, nullable=False)    # email = Column(String(32), unique=True)    # ctime = Column(DateTime, default=datetime.datetime.now)    # extra = Column(Text, nullable=True)    __table_args__ = (        # UniqueConstraint('id', 'name', name='uix_id_name'),        # Index('ix_id_name', 'name', 'email'),    )def init_db():    """    根据类创建数据库表    :return:     """    engine = create_engine(        "mysql+pymysql://root:123@127.0.0.1:3306/s6?charset=utf8",        max_overflow=0,  # 超过连接池大小外最多创建的连接        pool_size=5,  # 连接池大小        pool_timeout=30,  # 池中没有线程最多等待的时间，否则报错        pool_recycle=-1  # 多久之后对线程池中的线程进行一次连接的回收（重置）    )    Base.metadata.create_all(engine)def drop_db():    """    根据类删除数据库表    :return:     """    engine = create_engine(        "mysql+pymysql://root:123@127.0.0.1:3306/s6?charset=utf8",        max_overflow=0,  # 超过连接池大小外最多创建的连接        pool_size=5,  # 连接池大小        pool_timeout=30,  # 池中没有线程最多等待的时间，否则报错        pool_recycle=-1  # 多久之后对线程池中的线程进行一次连接的回收（重置）    )    Base.metadata.drop_all(engine)if __name__ == '__main__':    drop_db()    init_db()

创建单表

#!/usr/bin/env python# -*- coding:utf-8 -*-import datetimefrom sqlalchemy import create_enginefrom sqlalchemy.ext.declarative import declarative_basefrom sqlalchemy import Column, Integer, String, Text, ForeignKey, DateTime, UniqueConstraint, Indexfrom sqlalchemy.orm import relationshipBase = declarative_base()# ##################### 单表示例 #########################class Users(Base):    __tablename__ = 'users'    id = Column(Integer, primary_key=True)    name = Column(String(32), index=True)    age = Column(Integer, default=18)    email = Column(String(32), unique=True)    ctime = Column(DateTime, default=datetime.datetime.now)    extra = Column(Text, nullable=True)    __table_args__ = (        # UniqueConstraint('id', 'name', name='uix_id_name'),        # Index('ix_id_name', 'name', 'extra'),    )class Hosts(Base):    __tablename__ = 'hosts'    id = Column(Integer, primary_key=True)    name = Column(String(32), index=True)    ctime = Column(DateTime, default=datetime.datetime.now)# ##################### 一对多示例 #########################class Hobby(Base):    __tablename__ = 'hobby'    id = Column(Integer, primary_key=True)    caption = Column(String(50), default='篮球')class Person(Base):    __tablename__ = 'person'    nid = Column(Integer, primary_key=True)    name = Column(String(32), index=True, nullable=True)    hobby_id = Column(Integer, ForeignKey("hobby.id"))    # 与生成表结构无关，仅用于查询方便    hobby = relationship("Hobby", backref='pers')# ##################### 多对多示例 #########################class Server2Group(Base):    __tablename__ = 'server2group'    id = Column(Integer, primary_key=True, autoincrement=True)    server_id = Column(Integer, ForeignKey('server.id'))    group_id = Column(Integer, ForeignKey('group.id'))class Group(Base):    __tablename__ = 'group'    id = Column(Integer, primary_key=True)    name = Column(String(64), unique=True, nullable=False)    # 与生成表结构无关，仅用于查询方便    servers = relationship('Server', secondary='server2group', backref='groups')class Server(Base):    __tablename__ = 'server'    id = Column(Integer, primary_key=True, autoincrement=True)    hostname = Column(String(64), unique=True, nullable=False)def init_db():    """    根据类创建数据库表    :return:     """    engine = create_engine(        "mysql+pymysql://root:123@127.0.0.1:3306/s6?charset=utf8",        max_overflow=0,  # 超过连接池大小外最多创建的连接        pool_size=5,  # 连接池大小        pool_timeout=30,  # 池中没有线程最多等待的时间，否则报错        pool_recycle=-1  # 多久之后对线程池中的线程进行一次连接的回收（重置）    )    Base.metadata.create_all(engine)def drop_db():    """    根据类删除数据库表    :return:     """    engine = create_engine(        "mysql+pymysql://root:123@127.0.0.1:3306/s6?charset=utf8",        max_overflow=0,  # 超过连接池大小外最多创建的连接        pool_size=5,  # 连接池大小        pool_timeout=30,  # 池中没有线程最多等待的时间，否则报错        pool_recycle=-1  # 多久之后对线程池中的线程进行一次连接的回收（重置）    )    Base.metadata.drop_all(engine)if __name__ == '__main__':    drop_db()    init_db()

创建多个表并包含Fk、M2M关系

指定关联列：hobby = relationship("Hobby", backref='pers',foreign_keys="Person.hobby_id")

b. 操作数据库表

#!/usr/bin/env python# -*- coding:utf-8 -*-from sqlalchemy.orm import sessionmakerfrom sqlalchemy import create_enginefrom models import Users engine = create_engine("mysql+pymysql://root:123@127.0.0.1:3306/s6", max_overflow=0, pool_size=5)Session = sessionmaker(bind=engine) # 每次执行数据库操作时，都需要创建一个sessionsession = Session() # ############# 执行ORM操作 #############obj1 = Users(name="alex1")session.add(obj1) # 提交事务session.commit()# 关闭sessionsession.close()

#!/usr/bin/env python# -*- coding:utf-8 -*-from sqlalchemy.orm import sessionmakerfrom sqlalchemy import create_enginefrom sqlalchemy.orm import scoped_sessionfrom models import Usersengine = create_engine("mysql+pymysql://root:123@127.0.0.1:3306/s6", max_overflow=0, pool_size=5)Session = sessionmaker(bind=engine)"""# 线程安全，基于本地线程实现每个线程用同一个session# 特殊的：scoped_session中有原来方法的Session中的一下方法：public_methods = (    '__contains__', '__iter__', 'add', 'add_all', 'begin', 'begin_nested',    'close', 'commit', 'connection', 'delete', 'execute', 'expire',    'expire_all', 'expunge', 'expunge_all', 'flush', 'get_bind',    'is_modified', 'bulk_save_objects', 'bulk_insert_mappings',    'bulk_update_mappings',    'merge', 'query', 'refresh', 'rollback',    'scalar')"""session = scoped_session(Session)# ############# 执行ORM操作 #############obj1 = Users(name="alex1")session.add(obj1)# 提交事务session.commit()# 关闭sessionsession.close()

基于scoped_session实现线程安全

#!/usr/bin/env python# -*- coding:utf-8 -*-import timeimport threadingfrom sqlalchemy.ext.declarative import declarative_basefrom sqlalchemy import Column, Integer, String, ForeignKey, UniqueConstraint, Indexfrom sqlalchemy.orm import sessionmaker, relationshipfrom sqlalchemy import create_enginefrom db import Usersengine = create_engine("mysql+pymysql://root:123@127.0.0.1:3306/s6", max_overflow=0, pool_size=5)Session = sessionmaker(bind=engine)def task(arg):    session = Session()    obj1 = Users(name="alex1")    session.add(obj1)    session.commit()for i in range(10):    t = threading.Thread(target=task, args=(i,))    t.start()

多线程执行示例

#!/usr/bin/env python# -*- coding:utf-8 -*-import timeimport threadingfrom sqlalchemy.ext.declarative import declarative_basefrom sqlalchemy import Column, Integer, String, ForeignKey, UniqueConstraint, Indexfrom sqlalchemy.orm import sessionmaker, relationshipfrom sqlalchemy import create_enginefrom sqlalchemy.sql import textfrom db import Users, Hostsengine = create_engine("mysql+pymysql://root:123@127.0.0.1:3306/s6", max_overflow=0, pool_size=5)Session = sessionmaker(bind=engine)session = Session()# ################ 添加 ################"""obj1 = Users(name="wupeiqi")session.add(obj1)session.add_all([    Users(name="wupeiqi"),    Users(name="alex"),    Hosts(name="c1.com"),])session.commit()"""# ################ 删除 ################"""session.query(Users).filter(Users.id > 2).delete()session.commit()"""# ################ 修改 ################"""session.query(Users).filter(Users.id > 0).update({"name" : "099"})session.query(Users).filter(Users.id > 0).update({Users.name: Users.name + "099"}, synchronize_session=False)session.query(Users).filter(Users.id > 0).update({"age": Users.age + 1}, synchronize_session="evaluate")session.commit()"""# ################ 查询 ################"""r1 = session.query(Users).all()r2 = session.query(Users.name.label('xx'), Users.age).all()r3 = session.query(Users).filter(Users.name == "alex").all()r4 = session.query(Users).filter_by(name='alex').all()r5 = session.query(Users).filter_by(name='alex').first()r6 = session.query(Users).filter(text("id<:value and name=:name")).params(value=224, name='fred').order_by(Users.id).all()r7 = session.query(Users).from_statement(text("SELECT * FROM users where name=:name")).params(name='ed').all()"""session.close()

基本增删改查示例

#　条件ret = session.query(Users).filter_by(name='alex').all()ret = session.query(Users).filter(Users.id > 1, Users.name == 'eric').all()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()ret = session.query(Users).filter(~Users.id.in_([1,3,4])).all()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()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]# 排序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 funcret = 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()ret = session.query(Person).join(Favor).all()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()

常用操作

#!/usr/bin/env python# -*- coding:utf-8 -*-import timeimport threadingfrom sqlalchemy.ext.declarative import declarative_basefrom sqlalchemy import Column, Integer, String, ForeignKey, UniqueConstraint, Indexfrom sqlalchemy.orm import sessionmaker, relationshipfrom sqlalchemy import create_enginefrom sqlalchemy.sql import textfrom sqlalchemy.engine.result import ResultProxyfrom db import Users, Hostsengine = create_engine("mysql+pymysql://root:123@127.0.0.1:3306/s6", max_overflow=0, pool_size=5)Session = sessionmaker(bind=engine)session = Session()# 查询# cursor = session.execute('select * from users')# result = cursor.fetchall()# 添加cursor = session.execute('insert into users(name) values(:value)',params={
     "value":'wupeiqi'})session.commit()print(cursor.lastrowid)session.close()

原生SQL语句

#!/usr/bin/env python# -*- coding:utf-8 -*-import timeimport threadingfrom sqlalchemy.ext.declarative import declarative_basefrom sqlalchemy import Column, Integer, String, ForeignKey, UniqueConstraint, Indexfrom sqlalchemy.orm import sessionmaker, relationshipfrom sqlalchemy import create_enginefrom sqlalchemy.sql import textfrom sqlalchemy.engine.result import ResultProxyfrom db import Users, Hosts, Hobby, Personengine = create_engine("mysql+pymysql://root:123@127.0.0.1:3306/s6?charset=utf8", max_overflow=0, pool_size=5)Session = sessionmaker(bind=engine)session = Session()# 添加"""session.add_all([    Hobby(caption='乒乓球'),    Hobby(caption='羽毛球'),    Person(name='张三', hobby_id=3),    Person(name='李四', hobby_id=4),])person = Person(name='张九', hobby=Hobby(caption='姑娘'))session.add(person)hb = Hobby(caption='人妖')hb.pers = [Person(name='文飞'), Person(name='博雅')]session.add(hb)session.commit()"""# 使用relationship正向查询"""v = session.query(Person).first()print(v.name)print(v.hobby.caption)"""# 使用relationship反向查询"""v = session.query(Hobby).first()print(v.caption)print(v.pers)"""session.close()

基于relationship操作ForeignKey

#!/usr/bin/env python# -*- coding:utf-8 -*-import timeimport threadingfrom sqlalchemy.ext.declarative import declarative_basefrom sqlalchemy import Column, Integer, String, ForeignKey, UniqueConstraint, Indexfrom sqlalchemy.orm import sessionmaker, relationshipfrom sqlalchemy import create_enginefrom sqlalchemy.sql import textfrom sqlalchemy.engine.result import ResultProxyfrom db import Users, Hosts, Hobby, Person, Group, Server, Server2Groupengine = create_engine("mysql+pymysql://root:123@127.0.0.1:3306/s6?charset=utf8", max_overflow=0, pool_size=5)Session = sessionmaker(bind=engine)session = Session()# 添加"""session.add_all([    Server(hostname='c1.com'),    Server(hostname='c2.com'),    Group(name='A组'),    Group(name='B组'),])session.commit()s2g = Server2Group(server_id=1, group_id=1)session.add(s2g)session.commit()gp = Group(name='C组')gp.servers = [Server(hostname='c3.com'),Server(hostname='c4.com')]session.add(gp)session.commit()ser = Server(hostname='c6.com')ser.groups = [Group(name='F组'),Group(name='G组')]session.add(ser)session.commit()"""# 使用relationship正向查询"""v = session.query(Group).first()print(v.name)print(v.servers)"""# 使用relationship反向查询"""v = session.query(Server).first()print(v.hostname)print(v.groups)"""session.close()

基于relationship操作m2m

#!/usr/bin/env python# -*- coding:utf-8 -*-import timeimport threadingfrom sqlalchemy.ext.declarative import declarative_basefrom sqlalchemy import Column, Integer, String, ForeignKey, UniqueConstraint, Indexfrom sqlalchemy.orm import sessionmaker, relationshipfrom sqlalchemy import create_enginefrom sqlalchemy.sql import text, funcfrom sqlalchemy.engine.result import ResultProxyfrom db import Users, Hosts, Hobby, Person, Group, Server, Server2Groupengine = create_engine("mysql+pymysql://root:123@127.0.0.1:3306/s6?charset=utf8", max_overflow=0, pool_size=5)Session = sessionmaker(bind=engine)session = Session()# 关联子查询subqry = session.query(func.count(Server.id).label("sid")).filter(Server.id == Group.id).correlate(Group).as_scalar()result = session.query(Group.name, subqry)"""SELECT `group`.name AS group_name, (SELECT count(server.id) AS sid FROM server WHERE server.id = `group`.id) AS anon_1 FROM `group`"""# 原生SQL"""# 查询cursor = session.execute('select * from users')result = cursor.fetchall()# 添加cursor = session.execute('insert into users(name) values(:value)',params={"value":'wupeiqi'})session.commit()print(cursor.lastrowid)"""session.close()

其他

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

你可能感兴趣的文章