Python ORM框架之SQLAlchemy

時間 2019-11-10

標籤 python orm 框架 sqlalchemy 欄目 Python 简体版

原文原文鏈接

前言：

Django的ORM雖然強大，可是畢竟侷限在Django，而SQLAlchemy是Python中的ORM框架；html

SQLAlchemy的做用是：類/對象--->SQL語句--->經過pymysql/MySQLdb模塊--->提交到數據庫執行；

組成部分：python

Engine，框架的引擎
Connection Pooling ，數據庫鏈接池
Dialect，選擇鏈接數據庫的DB API種類
Schema/Types，架構和類型
SQL Exprression Language，SQL表達式語言

SQLAlchemy自己沒法操做數據庫，其必須以來pymsql等第三方插件，Dialect用於和數據API進行交流，根據配置文件的不一樣調用不一樣的數據庫API，從而實現對數據庫的操做，如：mysql

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

pip3 install sqlalchemy  #安裝sqlalchemy模塊

1、基本使用

1.原生SQLweb

import time
import threading
import sqlalchemy
from sqlalchemy import create_engine
from sqlalchemy.engine.base import Engine

conn_pool=create_engine( #建立sqlalchemy引擎
     "mysql+pymysql://webproject:web@192.168.1.18:3306/web?charset=utf8",
     max_overflow=2, #超過鏈接池大小以後，容許最大擴展鏈接數；
     pool_size=5,   #鏈接池大小
     pool_timeout=30,#鏈接池若是沒有鏈接了，最長等待時間
     pool_recycle=-1,#多久以後對鏈接池中鏈接進行一次回收

)


#單線程操做線程池

conn = conn_pool.raw_connection()  #從鏈接池中獲取1個鏈接,開始鏈接
cursor = conn.cursor()
cursor.execute(
    "select * from cmdb_worker_order"
)
result = cursor.fetchall()
print(result)
cursor.close()
conn.close()

單線程操做線程池

import time
import threading
import sqlalchemy
from sqlalchemy import create_engine
from sqlalchemy.engine.base import Engine

conn_pool=create_engine( #建立sqlalchemy引擎
     "mysql+pymysql://webproject:web@192.168.1.18:3306/web?charset=utf8",
     max_overflow=2, #超過鏈接池大小以後，容許最大擴展鏈接數；
     pool_size=5,   #鏈接池大小
     pool_timeout=30,#鏈接池若是沒有鏈接了，最長等待時間
     pool_recycle=-1,#多久以後對鏈接池中鏈接進行一次回收

)



#多線程操做線程池
def task(arg):
    conn = conn_pool.raw_connection()
    cursor = conn.cursor()
    cursor.execute(
        #"select * from cmdb_worker_order"
        "select sleep(2)"
    )
    result = cursor.fetchall()
    cursor.close()
    conn.close()


for i in range(20):
    t = threading.Thread(target=task, args=(i,)) #5個線程 執行2秒 而後5個線程在去執行2秒
    t.start()

多線程操做線程池

mysql> show status like 'Threads%';
+-------------------+-------+
| Variable_name     | Value |
+-------------------+-------+
| Threads_cached    | 1     |
| Threads_connected | 8     |
| Threads_created   | 11    |
| Threads_running   | 8     |
+-------------------+-------+
4 rows in set (0.00 sec)

2.ORMredis

2.1單表sql

a. 建立數據庫單表數據庫

#建立單表
#!/usr/bin/env python
# -*- coding:utf-8 -*-
import datetime
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String, Text, ForeignKey, DateTime, UniqueConstraint, Index

Base = declarative_base()


class Users(Base):
    __tablename__ = 'users'                #表名稱
    id = Column(Integer, primary_key=True) # primary_key=True設置主鍵
    name = Column(String(32), index=True, nullable=False) #index=True建立索引， nullable=False不爲空。

def init_db(): #根據類建立數據庫表
    engine = create_engine(
        "mysql+pymysql://webproject:web@192.168.1.18:3306/web?charset=utf8",
        max_overflow=0,   # 超過鏈接池大小外最多建立的鏈接
        pool_size=5,      # 鏈接池大小
        pool_timeout=30,  # 池中沒有線程最多等待的時間，不然報錯
        pool_recycle=-1   # 多久以後對線程池中的線程進行一次鏈接的回收（重置）
    )

    Base.metadata.create_all(engine) #這行代碼很關鍵哦！！ 讀取繼承了Base類的全部表在數據庫中進行建立

if __name__ == '__main__':
    init_db()                           #執行建立

建立單表

b.刪除表flask

#!/usr/bin/env python
# -*- coding:utf-8 -*-
import datetime
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String, Text, ForeignKey, DateTime, UniqueConstraint, Index

Base = declarative_base()


class Users(Base):
    __tablename__ = 'users'                #表名稱
    id = Column(Integer, primary_key=True) # primary_key=True設置主鍵
    name = Column(String(32), index=True, nullable=False) #index=True建立索引， nullable=False不爲空。

def drop_db(): #根據類 刪除數據庫表
    engine = create_engine(
        "mysql+pymysql://webproject:web@192.168.1.18:3306/web?charset=utf8",
        max_overflow=0,   # 超過鏈接池大小外最多建立的鏈接
        pool_size=5,      # 鏈接池大小
        pool_timeout=30,  # 池中沒有線程最多等待的時間，不然報錯
        pool_recycle=-1   # 多久以後對線程池中的線程進行一次鏈接的回收（重置）
    )

    Base.metadata.drop_all(engine) #這行代碼很關鍵哦！！ 讀取繼承了Base類的全部表在數據庫中進行刪除表

if __name__ == '__main__':
    drop_db()                          #執行建立

刪除表

c.添加1條記錄api

# import time
# import threading
# import sqlalchemy
# from sqlalchemy import create_engine
# from sqlalchemy.engine.base import Engine
# #
# conn_pool=create_engine( #建立sqlalchemy引擎
#      "mysql+pymysql://webproject:web@192.168.1.18:3306/web?charset=utf8",
#      max_overflow=2, #超過鏈接池大小以後，容許最大擴展鏈接數；
#      pool_size=5,   #鏈接池大小
#      pool_timeout=30,#鏈接池若是沒有鏈接了，最長等待時間
#      pool_recycle=-1,#多久以後對鏈接池中鏈接進行一次回收
#
# )

#
# #單線程操做線程池
#
# conn = conn_pool.raw_connection()  #從鏈接池中獲取1個鏈接,開始鏈接
# cursor = conn.cursor()
# cursor.execute(
#     "select * from cmdb_worker_order"
# )
# result = cursor.fetchall()
# print(result)
# cursor.close()
# conn.close()


#多線程操做線程池
# def task(arg):
#     conn = conn_pool.raw_connection()
#     cursor = conn.cursor()
#     cursor.execute(
#         #"select * from cmdb_worker_order"
#         "select sleep(2)"
#     )
#     result = cursor.fetchall()
#     cursor.close()
#     conn.close()
#
#
# for i in range(20):
#     t = threading.Thread(target=task, args=(i,)) #5個線程 執行2秒 而後5個線程在去執行2秒
#     t.start()
#
#


#建立單表
#!/usr/bin/env python
# -*- coding:utf-8 -*-
import datetime
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String, Text, ForeignKey, DateTime, UniqueConstraint, Index

Base = declarative_base()


class Users(Base):
    __tablename__ = 'users'                #表名稱
    id = Column(Integer, primary_key=True) # primary_key=True設置主鍵
    name = Column(String(32), index=True, nullable=False) #index=True建立索引， nullable=False不爲空。

def create_db(): #根據類 刪除數據庫表
    engine = create_engine(
        "mysql+pymysql://webproject:web@192.168.1.18:3306/web?charset=utf8",
        max_overflow=0,   # 超過鏈接池大小外最多建立的鏈接
        pool_size=5,      # 鏈接池大小
        pool_timeout=30,  # 池中沒有線程最多等待的時間，不然報錯
        pool_recycle=-1   # 多久以後對線程池中的線程進行一次鏈接的回收（重置）
    )

    Base.metadata.create_all(engine) #這行代碼很關鍵哦！！ 讀取繼承了Base類的全部表在數據庫中進行刪除表

if __name__ == '__main__':
    create_db()                          #執行建立

models.py

from SqlALchemy.models import Users
from sqlalchemy.orm import sessionmaker
from sqlalchemy import create_engine

engine = create_engine( "mysql+pymysql://webproject:web@192.168.1.18:3306/web?charset=utf8")
Session = sessionmaker(bind=engine)

# 每次執行數據庫操做時，都須要建立一個會話
session = Session()

# ############# 執行ORM操做 #############
obj1 = Users(name="張根") #建立Users對象=1行數據
session.add(obj1)          #添加到表中

# 提交事務
session.commit()
# 關閉session
session.close()

app01.py

2.2.多表session

#!/usr/bin/env python
# -*- coding:utf-8 -*-
import datetime
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String, Text, ForeignKey, DateTime, UniqueConstraint, Index
from sqlalchemy.orm import relationship

Base = declarative_base()

class Users(Base):
    __tablename__ = 'users'                #表名稱
    id = Column(Integer, primary_key=True) # primary_key=True設置主鍵
    name = Column(String(32), index=True, nullable=False) #index=True建立索引， nullable=False不爲空。
    age = Column(Integer, default=18)        #數字字段
    email = Column(String(32), unique=True)  #設置惟一索引
    ctime = Column(DateTime, default=datetime.datetime.now) #設置默認值爲當前時間（注意千萬不要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://webproject:web@192.168.1.18:3306/web?charset=utf8",
        max_overflow=0,  # 超過鏈接池大小外最多建立的鏈接
        pool_size=5,  # 鏈接池大小
        pool_timeout=30,  # 池中沒有線程最多等待的時間，不然報錯
        pool_recycle=-1  # 多久以後對線程池中的線程進行一次鏈接的回收（重置）
    )

    Base.metadata.create_all(engine)





def drop_db(): #根據類 刪除數據庫表
    engine = create_engine(
        "mysql+pymysql://webproject:web@192.168.1.18:3306/web?charset=utf8",
        max_overflow=0,   # 超過鏈接池大小外最多建立的鏈接
        pool_size=5,      # 鏈接池大小
        pool_timeout=30,  # 池中沒有線程最多等待的時間，不然報錯
        pool_recycle=-1   # 多久以後對線程池中的線程進行一次鏈接的回收（重置）
    )

    Base.metadata.drop_all(engine) #這行代碼很關鍵哦！！ 讀取繼承了Base類的全部表在數據庫中進行刪除表

if __name__ == '__main__':
    init_db()                      #執行建立

建立多表

2、源碼剖析

使用scoped_session(Session) 和Session() 建立的鏈接的區別？

A.發現問題：

爲何 session =scoped_session(Session) 和 session = Session()，明明是2個沒有繼承關係的類實例化出來的對象，卻有相同的add/commit/...方法？

class A(object):
    fields=('f1','f2' )
    def f1(self):
        print('A類的f1方法')
    def f2(self):
        print('A的f2方法')

    def __call__(self,*args, **kwargs):
        getattr(self,*args)()

class B(object):
    def __init__(self,class_A):
        self.class_a=class_A()


def waper(name):
    def do(self):
       self.class_a(name)
    return do


for name in A.fields:
    setattr(B,name,waper(name))


obj=B(A)
obj.f1()
obj.f2()

源碼流程分析僞代碼

B.session =scoped_session(Session) 實例化執行scoped_session的__init__方法，Session參數也就是原session類；

class scoped_session(object): #沒有繼承
    session_factory = None
    def __init__(self, session_factory, scopefunc=None):
        self.session_factory = session_factory   #1.0 ：session_factory=原來的session類
        if scopefunc:                           # 1.1：scopefunc=None 走else分支
            self.registry = ScopedRegistry(session_factory, scopefunc)
        else:
            '''
       class ThreadLocalRegistry(ScopedRegistry):
            def __init__(self, createfunc):
                self.createfunc = createfunc        #源session類
                self.registry = threading.local()   #封裝了1個能夠隔離多線程之間數據的threading.local()對象：  
          '''
            self.registry = ThreadLocalRegistry(session_factory)  #返回1個封裝了源session類和threading.local對象的ThreadLocalRegistry對象

C.給 scoped_session類設置屬性 = 1個封裝了閉包函數do，封裝了這些屬性，在用戶app裏實例化 scoped_session()以後就能夠去執行這些do函數了！

def instrument(name):

    def do(self, *args, **kwargs): #self=scoped_session對象

        return getattr(self.registry(), name)(*args, **kwargs) #self    name=add /commit 閉包封裝進來的
        '''
        把一下代碼封裝到 scoped_session類中去，接下若是執行self就是scoped_session對象 或者ScopedRegistry對象了

了！ self.session_factory = session_factory #session_factory=原來的session類 self.registry = ScopedRegistry(session_factory, scopefunc) #ScopedRegistry對象 name: def do(self, *args, **kwargs): return getattr(self.registry(), add )(*args, **kwargs) ''' return do

D.app中執行session.add(obj1)本質是執行scoped_session類中封裝的add屬性對應的do函數

    def do(self, *args, **kwargs): 
        # self.registry()=執行ThreadLocalRegistry 或者 scoped_session對象 的__call__方法 
        return getattr(self.registry(), name)(*args, **kwargs) #self    name=add /commit 閉包封裝進來的



#最後執行下面的代碼！

def __call__(self):
    try:
        return self.registry.value #去threading.local()獲取
    except AttributeError:        #若是獲取不到
        val = self.registry.value = self.createfunc() #去源session對象中獲取方法
        return val

E.得出結論：

scoped_session(Session) 內部使用了threading.local() 實現了對多線程的支持；

F.知識：

__all__ = ['scoped_session'] ：1個py文件中使用了__all__=[ ]限制了導入的變量；

threading.local()：爲每1個線程，另外開闢1塊新內存，來保存local_value，因此每一個線程均可以獲取到本身設置的值。

閉包：能夠把外部函數數據，傳遞到內部函數中保存；

3、進階操做

注意不管SQLalchemy的增、刪、改、查操做，最後都須要commit，數據庫纔會執行SQL；

1.增、刪、改操做

obj1 = Users(name="張根",age=18,email='13220198866@163.com',extra='sss')
session.add(obj1)

單條增長

session.add_all([
Users(name="張根1",age=19,email='645172205@qq.com',extra='sss'),
Users(name="張根2",age=20,email='13220198866@139.com',extra='sss')

])

批量增長

session.query(Users).filter(Users.id==5).delete()

刪除操做

###################修改##########################
session.query(Users).filter(Users.id==4).update({'name':'Martin'})
session.query(Users).filter(Users.id==4).update({Users.name: Users.name + "666"}, synchronize_session=False)#在原來的基礎上作字符串
session.query(Users).filter(Users.id==18).update({Users.id: Users.id - 12},synchronize_session="evaluate")#在原來的基礎上作數字+,-操做

修改操做

2.單表查詢操做

r0=session.query(Users).all()                             #查詢user表中所有數據；
r1=session.query(Users).filter(Users.id > 2)              #查詢user表中，id>2的記錄；
r2=session.query(Users.age).all()                         #查詢User表中的 age列； ##[(18,), (19,)]
r3=session.query(Users.age.label('cname')).all()          #使用別名查詢

r4=session.query(Users).filter(Users.name=='Martin').all()      #查詢姓名==Martin的用戶
r5=session.query(Users).filter_by(name='Martin',age=19).all()   #filter_by方式查詢
r6=session.query(Users).filter_by(name='Martin',age=19).first() #獲取第 單個對象 print(r6.name)
r7=session.query(Users).filter(text("id<:value and name=:name")).params(value=18, name='Martin').order_by(Users.id).all()
                                                                    #查詢 id>18 name=Martin的Users 根據 id排序，params支持傳參數；
r8 = session.query(Users).from_statement(text("SELECT * FROM users where name=:name")).params(name='ed').all()#from_statement,申請使用原生sql

基本查詢操做

#條件查詢
ret0 = session.query(Users).filter(Users.id.between(0,7), Users.name == 'Martin').all()
# #between: 查詢 user id在0--7之間，用戶名爲Martin 的數據;
ret1= session.query(Users).filter(Users.id.in_([1,6])).all()
#查詢user_id in [1,6] 的數據
ret2 = session.query(Users).filter(~Users.id.in_([1,3,4])).all()
#查詢user_id  not in [1,6] 的數據

ret = session.query(Users).filter(Users.id.in_(session.query(Users.id).filter_by(name='Martin'))).all()
#1.session.query(Users.id).filter_by(name='Martin') 查詢name=Martin'的id
#2.在user表中 按1的結果 查詢

條件查詢

#################################邏輯運算#####################################

from sqlalchemy import and_, or_
ret0 = session.query(Users).filter(and_(Users.id > 3, Users.name == 'eric')).all()
ret1 = session.query(Users).filter(or_(Users.id < 2, Users.name == 'eric')).all()
ret2 = session.query(Users).filter(
    or_(
        Users.id < 2,
        and_(Users.name == 'eric', Users.id > 3),
        Users.extra != ""
    )).all()

多查詢條件邏輯運算、嵌套查詢

###################################字符串符 模糊匹配查詢###############################

ret0 = session.query(Users).filter(Users.name.like('M%')).first().name
ret1 = session.query(Users).filter(~Users.name.like('%r%')).first().name
print(ret0,ret1)

字符串符模糊匹配查詢

########################### 限制（分頁）############################
ret = session.query(Users)[0:2]
print(ret)

限制分頁

##########################排序##############################
ret0 = session.query(Users).order_by(Users.id.desc()).all()                   #根據id,由大到小排序(desc).
ret1 = session.query(Users).order_by(Users.id.asc(),Users.age.desc()).all()   #根據id,由小到大排序(asc)，如id相等,由大到小排序(desc).
print([i.id for i in ret0 ]  )
print([i.id for i in ret1 ]  )

排序

################################分組###############################

from sqlalchemy.sql import func

ret0 = session.query(Users).group_by(Users.age).all() #根據name字段進行分組

ret1 = session.query(     #使用name字段進行分組，求每組中 最大id 、最小id 、id 之和
    func.max(Users.id),
    func.sum(Users.id),
    func.min(Users.id)) .group_by(Users.name).all()


ret2 = session.query(
    func.max(Users.id),
    func.sum(Users.id),                          #對分組以後的數據進行 having篩選，
    func.min(Users.id)).group_by(Users.name).having(func.min(Users.id) >2).all()
'''
having的做用：
例如：查詢公司中 部門人數大於3人的部門，先按照部門分組，而後求人數，而後再having 大於3的；

'''

group_by分組查詢和.having二次篩選

##################################連表查詢###############################################

ret = session.query(models.Users).join(models.Hobby,models.Users.id == models.Hobby.id,isouter=True).filter(models.Users.id >1)
#2個沒有外鍵關係的表 作連表查詢
print(ret)

ret = session.query(models.Person).join(models.Hobby).all()              #inin_join
print(ret)
ret = session.query(models.Person).join(models.Hobby,isouter=True).all() #left_join 調換位置 更換爲 right_join
print(ret)

''''
left join：以左表爲準，顯示符合搜索條件的記錄；

aID　　　　　aNum　　　　　bID　　　　　bName
5　　　　　a20050115　　　　NULL　　　　　NULL


right join：以右表爲準，顯示符合搜索條件的記錄；

aID　　　　　aNum　　　　　bID　　　　　bName
NULL　　　　　NULL　　　　　8　　　　　2006032408


inin_join：並不以誰爲基礎,它只顯示符合條件
aID　　　　　aNum　　　　　bID　　　　　bName


'''

連表查詢

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

組合

3.基於relationship的增長、查詢操做

relationship能夠幫助咱們快速在存在1對多、多對多關係的表之間作反向連表查詢和數據添加；

#連表方式1：指定字段獲取
persons=session.query(models.Person.name,models.Hobby.caption.label('hobby_caption')).join(models.Hobby)
for row in  persons:
    print(row.name,row.hobby_caption,)

#連表方式2：獲取全部字段
persons = session.query(models.Person, models.Hobby).join(models.Hobby)
for row in persons:
    #print(row)=2個對象
    print(row[0].name,row[1].caption)

#連表方式3：relationship 連表查詢
persons = session.query(models.Person).all()
for row in persons:
    print(row.name,row.hobby.caption)

'''
 hobby = relationship("Hobby", backref='pers')
 Hobby：正向查詢
 backref：反向查詢

ps:查詢喜歡姑娘的全部人
hobby_obj=session.query(models.Hobby).filter(models.Hobby.id==2).first()
print(hobby_obj.pers)
'''

################################relationship增長######################
#1.relationship正向增長
person_obj=models.Person(name='Tony',hobby=models.Hobby(caption='any'))


#2.relationship 反向增長
hobby_obj=models.Hobby(caption='人妖')
hobby_obj.pers=[
        models.Person(name='李淵'),
        models.Person(name='西門慶'),
    ]
session.add(person_obj,hobby_obj)

基於relationship 作1對多操做

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
from sqlalchemy.sql import text
from sqlalchemy.engine.result import ResultProxy
from SqlALchemy import models

engine = create_engine( "mysql+pymysql://webproject:web@192.168.1.18:3306/web?charset=utf8")
Session = sessionmaker(bind=engine)
session = Session()

#############################多對多添加#################################

#添加方式1：同時添加男、女對象，直接添加相親表；前提是 知道新添加男、女對象的ID；
session.add_all(
    [
    models.Boy(name='張無忌'),
    models.Boy(name='宋青書'),
    models.Girl(name='周芷若'),
    models.Girl(name='趙敏'),
])
session.commit()

s2g =models.G2B(girl_id=1,boy_id=1)
session.add(s2g)
session.commit()

#添加方式2：經過女生對象 添加 相親記錄表
girl_obj = models.Girl(name='滅絕師太')                                     #建立1位女性朋友 滅絕
girl_obj.boys = [models.Boy(name='張三丰'),models.Boy(name='方正大師')]    #而後滅絕和 張三丰、方正大師分別相了1次親
session.add(girl_obj)
session.commit()

##添加方式3：經過男生對象 添加 相親記錄
boy_obj = session.query(models.Boy).filter(models.Boy.name=='尹志平').first()  #建立1位男性朋友 尹志平
boy_obj.girls = [models.Girl(name='小龍女'),models.Girl(name='黃蓉')]         #而後尹志平 和小龍女、黃蓉分別 相了一次親
session.add(boy_obj)
session.commit()

##################################多對多查詢###################################

#boys = relationship('Boy', secondary='g2b', backref='girls')
#1.基於 relationship 正向查詢
mie_jue = session.query(models.Girl).filter(models.Girl.name=='滅絕師太').first()
print( [i.name for i in mie_jue.boys]) #['方正大師', '張三丰']

#2.基於 relationship 反向查詢
yi_zhi_ping = session.query(models.Boy).filter(models.Boy.name=='尹志平').first()
print( [i.name for i in yi_zhi_ping.girls]) #['小龍女', '黃蓉']

基於relationship 多對多操做

如下爲相親表表結構：

###################### 相親表多對多示例 #########################
class G2B(Base):
    __tablename__ = 'g2b'
    id = Column(Integer, primary_key=True, autoincrement=True)
    girl_id = Column(Integer, ForeignKey('girl.id'))
    boy_id = Column(Integer, ForeignKey('boy.id'))

class Girl(Base):
    __tablename__ = 'girl'
    id = Column(Integer, primary_key=True)
    name = Column(String(64), unique=True, nullable=False)

    # 與生成表結構無關，僅用於查詢方便
    boys = relationship('Boy', secondary='g2b', backref='girls')


class Boy(Base):
    __tablename__ = 'boy'
    id = Column(Integer, primary_key=True, autoincrement=True)
    name = Column(String(64), unique=True, nullable=False)

models

4.關聯子查詢

什麼是SQL子查詢？

mysql> select id,name,(select max(id) from girl) as maxgirl from boy;      #SQL子查詢 +----+--------------+---------+
| id | name         | maxgirl |
+----+--------------+---------+
|  2 | 宋青書       |       7 |
|  5 | 尹志平       |       7 |
|  3 | 張三丰       |       7 |
|  1 | 張無忌       |       7 |
|  4 | 方正大師     |       7 |
+----+--------------+---------+
5 rows in set (0.00 sec)

mysql>

查詢每一個學生的平均分！

First, query the SID from Student

Second,with SID query everyone `s socres compute average score。

select id,name，(select avg(score) from 成績表 where 成績表.sid =學生表.id ) as avg_socre from 學生表；

#!/usr/bin/env python
# -*- coding:utf-8 -*-
import time
import threading

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
from sqlalchemy.sql import text, func
from sqlalchemy.engine.result import ResultProxy
from db import Users, Hosts, Hobby, Person, Group, Server, Server2Group

engine = 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()

SQLALchemy關聯子查詢操做

4、Flask-SQLAlchemy組件

FlaskSQLAlchemy是flask和SQLALchemy的管理者，其本質是在flask項目中經過對文件管理、導入，把Flask和QLAlchemy兩個組件無縫鏈接在一塊兒，

還能夠幫助咱們實現自動開啓、關閉鏈接、配置提高開發效率；

根據一個常見flask項目的目錄結構，梳理一下它的運行流程便知；

sansa項目

程序入口run.py導入sansa包執行__init__.py文件

0.導入sansa包會執行sanas的__init__.py文件導入create_app

1.執行create_app函數

#!/usr/bin/env python
# -*- coding:utf-8 -*-
"""
生成依賴文件：
    pipreqs ./

"""
from sansa import create_app  #0.導入sansa包會執行sanas的__init__.py文件導入create_app

app = create_app()              #1.執行create_app函數

if __name__ == '__main__':
    app.run()

run.py

執行sansa.__init__.py

0.導入flask_sqlalchemy,注意這裏導入的是flask_sqlalchemy不是原始的sqlalchemy

1.讀取、註冊flask的配置文件

2.經過配置文件，將flask_sqlalchemy註冊到app中

3.經過flask藍圖把account.account(路由和視圖) 註冊到app裏(導入視圖)

#!/usr/bin/env python
# -*- coding:utf-8 -*-
from flask import Flask
#0、導入flask_sqlalchemy,注意這裏導入的是flask_sqlalchemy不是原始的sqlalchemy
from flask_sqlalchemy import SQLAlchemy
db = SQLAlchemy()

from .models import *
from .views import account

def create_app():
    app = Flask(__name__)
    #一、讀取、註冊flask的配置文件
    app.config.from_object('settings.DevelopmentConfig')
    #二、經過配置文件，將flask_sqlalchemy註冊到app中
    db.init_app(app)
    #三、經過藍圖把account.account(路由和視圖) 註冊到app裏
    app.register_blueprint(account.account)  #導入視圖
    return app

項目\__init__.py

db對象在執行run.py剛剛啓動調用了sansa\__init__.py程序的時候就實例化好了並封裝好了配置文件、self.Model = self.make_declarative_base(model_class, metadata)如今就可使用了db對象建立models文件了。

開始建立model

0.db在執行run.py剛剛啓動調用了sansa\__init__.py程序的時候就實例化好了

1.導入sansa.__init__中的實例化完成的db對象class Users(db.Model):

2.db對象封裝好了配置信息、ORM基類、create_all方法

#!/usr/bin/env python
# -*- coding:utf-8 -*      #0.db在執行run.py剛剛啓動調用了sansa\__init__.py程序的時候就實例化好了
from . import db         #1.導入sansa.__init__中的實例化完成的db對象
class Users(db.Model):    #2.db在啓動的時候 已經封裝好了 配置文件、self.Model = self.make_declarative_base(model_class, metadata) 就可使用了
    """
    用戶表
    """
    __tablename__ = 'users'
    id = db.Column(db.Integer, primary_key=True)
    username = db.Column(db.String(80), unique=True, nullable=False)
    email = db.Column(db.String(120), unique=True, nullable=False)

    def __repr__(self):
        return '<User %r>' % self.username

model.py

讀取models.py中的映射去執行SQL建立表

0.加載models表映射關係

1.建立app對象

2.使用db對象根據model這種映射關係執行建立表操做

#!/usr/bin/env python
# -*- coding:utf-8 -*-
from sansa import create_app
from sansa import db
app = create_app()
from sansa.models import * #0.加載models表映射關係
with app.app_context():     #1.建立app對象
    db.create_all()          #2.使用db對象根據model這種映射關係執行建立表操做

create_table.py

經過視圖操做表

0.導入db對象，包含了engin和建立鏈接；

1.導入models；

2. db.session直接獲取鏈接，開始操做。。。。

#!/usr/bin/env python
# -*- coding:utf-8 -*-
from sansa import create_app
from sansa import db
app = create_app()
from sansa.models import * #0.加載models表映射關係
with app.app_context():     #1.建立app對象
    db.create_all()          #2.使用db對象根據model這種映射關係執行建立表操做
    #db.drop_all()           #3.使用db對象 刪除表

views\account.py

5、pipreqs組件

拿到別人的新項目以後發現缺乏這個、那個....模塊運行不起來，而後根據報錯逐一得去pip到最後發現安裝得版本不一致；

這不是你的問題而是項目開發者的不夠規範；

1.安裝pipreqs組件

pip install pipreqs

2.在項目/目錄下執行pipreqs ./，蒐集項目中全部使用得第三方包；

[root@cmdb cmdb_rbac_arya]# pipreqs ./
INFO: Successfully saved requirements file in ./requirements.txt
[root@cmdb cmdb_rbac_arya]# ls
123.txt                cron_ansible_api.py  manage.py   requirements.txt  webcron
ansible_api_runner.py  cron_close_order.py  Monaco.ttf  static            web.sql
arya                   cron_writesql.py     multitask   templates         work_order_remind.py
cmdb                   DBshow               nohup.out   tools
cmdb_rbac_arya         Get_biying_image.py  rbac        w8.pid
[root@cmdb cmdb_rbac_arya]# cat requirements.txt 
paramiko==2.4.1
ansible==2.6.3
PyMySQL==0.8.0
pandas==0.22.0
Django==1.11.7
XlsxWriter==1.0.4
redis==2.10.6
requests==2.18.4
Pillow==5.3.0
[root@cmdb cmdb_rbac_arya]#