Flask11--Flask-script，sqlalchemy

一.flask-script

用於實現類似於django中 python3 manage.py runserver ...類似的命令

安裝：pip3 install flask-script

1.1使用

from flask_script import Manager
app = Flask(__name__)
manager=Manager(app)
...
if __name__ == '__main__':
    manager.run()
#以后在執行，直接：python3 manage.py runserver
#python3 manage.py runserver --help

1.2自定制命令

@manager.command
def custom(arg):
    """
    自定義命令
    python manage.py custom 123
    :param arg:
    :return:
    """
    print(arg)
@manager.option('-n', '--name', dest='name')
#@manager.option('-u', '--url', dest='url')
def cmd(name, url):
    """
    自定義命令（-n也可以寫成--name）
    執行： python manage.py  cmd -n lqz -u http://www.oldboyedu.com
    執行： python manage.py  cmd --name lqz --url http://www.oldboyedu.com
    :param name:
    :param url:
    :return:
    """
    print(name, url)
#有什么用？

 

二. SQLAlchemy

1.介紹

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://<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

django中如何反向生成models

python manage.py inspectdb > app/models.py

2.簡單使用（能創建表，刪除表，不能修改表）

修改表：在數據庫添加字段，類對應上

1執行原生sql（不常用）

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

engine = create_engine(
    "mysql+pymysql://root:123456@127.0.0.1:3306/test?charset=utf8",
    max_overflow=0,  # 超過連接池大小外最多創建的連接
    pool_size=5,  # 連接池大小
    pool_timeout=30,  # 池中沒有線程最多等待的時間，否則報錯
    pool_recycle=-1  # 多久之后對線程池中的線程進行一次連接的回收（重置）
)

conn = engine.raw_connection()
cursor = conn.cursor()
cursor.execute("select * from app01_book")
result = cursor.fetchall()
print(result)
cursor.close()
conn.close()

2 orm使用

models.py

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()  #繼承的模型  類似於django的Model


class Users(Base):
    __tablename__ = 'users'  # 數據庫表名稱
    id = Column(Integer, primary_key=True)  # id 主鍵
    name = Column(String(32), index=True, nullable=False)  # name列，索引，不可為空
    # 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:123456@127.0.0.1:3306/aaa?charset=utf8",
        max_overflow=0,  # 超過連接池大小外最多創建的連接
        pool_size=5,  # 連接池大小
        pool_timeout=30,  # 池中沒有線程最多等待的時間，否則報錯
        pool_recycle=-1  # 多久之后對線程池中的線程進行一次連接的回收（重置）
    )

    Base.metadata.create_all(engine)

    # Base.metadata.drop_all(engine)   顯而易見 是刪除數據庫

if __name__ == '__main__':
    # drop_db()
    init_db()

app.py

from sqlalchemy.orm import sessionmaker
from sqlalchemy import create_engine
from models import Users
#"mysql+pymysql://root@127.0.0.1:3306/aaa"
engine = create_engine("mysql+pymysql://root:123456@127.0.0.1:3306/aaa", max_overflow=0, pool_size=5)
Connection = sessionmaker(bind=engine)
# 每次執行數據庫操作時，都需要創建一個Connection
con = Connection()
# ############# 執行ORM操作 #############
obj1 = Users(name="lqz")
con.add(obj1)
# 提交事務
con.commit()
# 關閉session，其實是將連接放回連接池
con.close()

3.一對多關系

Copyclass 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指的是tablename而不是類名，uselist=False
    hobby_id = Column(Integer, ForeignKey("hobby.id"))
    
    # 跟數據庫無關，不會新增字段，只用於快速鏈表操作
    # 類名，backref用於反向查詢
    hobby=relationship('Hobby',backref='pers')

4.多對多關系

class Boy2Girl(Base):
    __tablename__ = 'boy2girl'
    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)

class Boy(Base):
    __tablename__ = 'boy'

    id = Column(Integer, primary_key=True, autoincrement=True)
    hostname = Column(String(64), unique=True, nullable=False)
    
    # 與生成表結構無關，僅用於查詢方便,放在哪個單表中都可以
    servers = relationship('Girl', secondary='boy2girl', backref='boys')

5.操作數據表

from sqlalchemy.orm import sessionmaker
from sqlalchemy import create_engine
from models import Users
  
engine = create_engine("mysql+pymysql://root:123456@127.0.0.1:3306/aaa", max_overflow=0, pool_size=5)


Session = sessionmaker(bind=engine)
  
# 每次執行數據庫操作時，都需要創建一個session
session = Session()
  
# ############# 執行ORM操作 #############
obj1 = Users(name="lqz")
session.add(obj1)
  
# 提交事務
session.commit()
# 關閉session
session.close()

6.基於scoped_session實現線程安全

Copyfrom sqlalchemy.orm import sessionmaker
from sqlalchemy import create_engine
from sqlalchemy.orm import scoped_session
from 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)

"""
# 線程安全，基於本地線程實現每個線程用同一個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'
)
"""
#scoped_session類並沒有繼承Session,但是卻又它的所有方法
session = scoped_session(Session)
# ############# 執行ORM操作 #############
obj1 = Users(name="alex1")
session.add(obj1)

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

7.基本增刪查改

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
from db import Users, Hosts

engine = create_engine("mysql+pymysql://root:123@127.0.0.1:3306/s6", max_overflow=0, pool_size=5)
Session = sessionmaker(bind=engine)

session = Session()

----------------1.增-----------------

obj1 = Users(name="111")
session.add(obj1)

session.add_all([
    Users(name="111"),
    Users(name="222"),
    Hosts(name="333"),
])
session.commit()
----------------2.刪除-----------------
session.query(Users).filter(Users.id > 2).delete()
session.commit()
----------------3.修改-----------------
#傳字典
1.  session.query(Users).filter(Users.id > 0).update({"name" : "lqz"})


2.  session.query(Users).filter(Users.id > 0).update({Users.name: Users.name + "099"}, synchronize_session=False)  # 字符串  類似於django的F查詢

3.  session.query(Users).filter(Users.id > 0).update({"age": Users.age + 1}, 							synchronize_session="evaluate")   #數字
session.commit()

----------------4.查-----------------

1.r1 = session.query(Users).all()


2. r2 = session.query(Users.name.label('xx'), Users.age).all() #只取age列，把name																重命名為xx
3.#filter傳的是表達式，filter_by傳的是參數
    r3 = session.query(Users).filter(Users.name == "lqz").all()
    r4 = session.query(Users).filter_by(name='lqz').all()
    r5 = session.query(Users).filter_by(name='lqz').first()

4.#:value 和:name 相當於占位符，用params傳參數
    r6 = session.query(Users).filter(text("id<:value and name=:name")).params(value=224, name='fred').order_by(Users.id).all()
    
    
 5.#自定義查詢sql
r7 = session.query(Users).from_statement(text("SELECT * FROM users where name=:name")).params(name='ed').all()


#增，刪，改都要commit()
session.close()

8 高級操作

from sqlalchemy.orm import sessionmaker
from sqlalchemy import create_engine
from models import User,Person,Hobby
from sqlalchemy.sql import text
engine = create_engine("mysql+pymysql://root:123@127.0.0.1:3306/aaa", max_overflow=0, pool_size=5)
Session = sessionmaker(bind=engine)
session=Session()


# 1 查詢名字為lqz的所有user對象
# ret = session.query(User).filter_by(name='ccc099').all()
# 2 表達式，and條件連接
# ret = session.query(User).filter(User.id > 1, User.name == 'egon').all()
# 查找id在1和10之間，並且name=egon的對象
# ret = session.query(User).filter(User.id.between(1, 10), User.name == 'egon').all()

# in條件(class_,因為這是關鍵字，不能直接用)
# ret = session.query(User).filter(User.id.in_([1,3,4])).all()

# 取反 ~
ret = session.query(User).filter(~User.id.in_([1,3,4])).all()

#二次篩選
# select *
# ret = session.query(User).filter(User.id.in_(session.query(User.id).filter_by(name='egon'))).all()
# # select name，id 。。。。
# ret = session.query(User.id,User.name).filter(User.id.in_(session.query(User.id).filter_by(name='egon'))).all()

'''
SELECT users.id AS users_id, users.name AS users_name 
FROM users 
WHERE users.id IN (SELECT users.id AS users_id 
FROM users 
WHERE users.name = %(name_1)s)

'''


#
from sqlalchemy import and_, or_
#or_包裹的都是or條件，and_包裹的都是and條件
#查詢id>3並且name=egon的人
# ret = session.query(User).filter(and_(User.id > 3, User.name == 'egon')).all()

# 查詢id大於2或者name=ccc099的數據
# ret = session.query(User).filter(or_(User.id > 2, User.name == 'ccc099')).all()
# ret = session.query(User).filter(
#     or_(
#         User.id < 2,
#         and_(User.name == 'egon', User.id > 3),
#         User.extra != ""
#     )).all()
# print(ret)

'''
select *from user where id<2 or (name=egon and id >3) or extra !=''
'''


# 通配符，以e開頭，不以e開頭
# ret = session.query(User).filter(User.name.like('e%')).all()   
# ret = session.query(User).filter(~User.name.like('e%')).all()
											  like('S_')  匹配一個
    											%	 多個					

# 限制，用於分頁，區間 limit
# 前閉后開區間，1能取到，3取不到
ret = session.query(User)[1:3]

'''
select * from users limit 1,2;
'''


# 排序，根據name降序排列（從大到小）
# ret = session.query(User).order_by(User.name.desc()).all()
# ret = session.query(User).order_by(User.name.asc()).all()
#第一個條件降序排序后，再按第二個條件升序排
# ret = session.query(User).order_by(User.id.asc(),User.name.desc()).all()
# ret = session.query(User).order_by(User.name.desc(),User.id.asc()).all()


# 分組
from sqlalchemy.sql import func

# ret = session.query(User).group_by(User.name).all()
#分組之后取最大id，id之和，最小id
# sql 分組之后，要查詢的字段只能有分組字段和聚合函數
# ret = session.query(
#     func.max(User.id),
#     func.sum(User.id),
#     func.min(User.id),
#     User.name).group_by(User.name).all()
# '''
# select max(id),sum(id),min(id) from user group by name;
#
# '''
# for obj in ret:
#     print(obj[0],'----',obj[1],'-----',obj[2],'-----',obj[3])
# print(ret)

#haviing篩選
# ret = session.query(
#     func.max(User.id),
#     func.sum(User.id),
#     func.min(User.id)).group_by(User.name).having(func.min(User.id) >2).all()

'''
select max(id),sum(id),min(id) from user group by name having min(id)>2;

'''
print(ret)
session.commit()

session.close()

9.一對多

from sqlalchemy.orm import sessionmaker
from sqlalchemy import create_engine
from models import User,Person,Hobby,Boy,Girl,Boy2Girl
from sqlalchemy.sql import text
engine = create_engine("mysql+pymysql://root:123@127.0.0.1:3306/aaa", max_overflow=0, pool_size=5)
Session = sessionmaker(bind=engine)
session=Session()

###  1 一對多插入數據
obj=Hobby(caption='足球')
session.add(obj)
p=Person(name='張三',hobby_id=2)
session.add(p)
### 2 方式二(默認情況傳對象有問題)
###### Person表中要加 hobby = relationship('Hobby', backref='pers')
p=Person(name='李四',hobby=Hobby(caption='美女'))
等同於
p=Person(name='李四2')
p.hobby=Hobby(caption='美女2')
session.add(p)

## 3 方式三，通過反向操作
hb = Hobby(caption='人妖')
hb.pers = [Person(name='文飛'), Person(name='博雅')]
session.add(hb)


4 查詢（查詢：基於連表的查詢，基於對象的跨表查詢）
### 4.1 基於對象的跨表查詢(子查詢，兩次查詢)
 正查
p=session.query(Person).filter_by(name='張三').first()
print(p)
print(p.hobby.caption)
# 反查
h=session.query(Hobby).filter_by(caption='人妖').first()
print(h.pers)

### 4.2 基於連表的跨表查（查一次）
默認根據外鍵連表
isouter=True 左外連，表示Person left join Hobby，沒有右連接，反過來即可
# 不寫 inner join
person_list=session.query(Person,Hobby).join(Hobby,isouter=True).all()
print(person_list)
print(person_list)
for row in person_list:
print(row[0].name,row[1].caption)

 '''
 select * from person left join hobby on person.hobby_id=hobby.id

'''

ret = session.query(Person, Hobby).filter(Person.hobby_id == Hobby.id)
print(ret)
 '''
 select * from user,hobby where user.id=favor.nid;
 
 '''


#join表，默認是inner join
# ret = session.query(Person).join(Hobby)
# # ret = session.query(Hobby).join(Person,isouter=True)
# '''
# SELECT *
# FROM person INNER JOIN hobby ON hobby.id = person.hobby_id
# '''
# print(ret)


# 指定連表字段（從來沒用過）
ret = session.query(Person).join(Hobby,Person.nid==Hobby.id, isouter=True)
ret = session.query(Person).join(Hobby,Person.hobby_id==Hobby.id, isouter=True).all()
# print(ret)
'''
SELECT *
FROM person LEFT OUTER JOIN hobby ON person.nid = hobby.id

'''

# print(ret)

# 組合（了解）UNION 操作符用於合並兩個或多個 SELECT 語句的結果集
union和union all的區別？
q1 = session.query(User.name).filter(User.id > 2)  # 6條數據
q2 = session.query(User.name).filter(User.id < 8) # 2條數據


q1 = session.query(User.id,User.name).filter(User.id > 2)  # 6條數據
q2 = session.query(User.id,User.name).filter(User.id < 8) # 2條數據
ret = q1.union_all(q2).all()
ret1 = q1.union(q2).all()
print(ret)
print(ret1)

q1 = session.query(User.name).filter(User.id > 2)
q2 = session.query(Hobby.caption).filter(Hobby.nid < 2)
ret = q1.union_all(q2).all()

10.多對多

# session.add_all([
#     Boy(hostname='霍建華'),
#     Boy(hostname='胡歌'),
#     Girl(name='劉亦菲'),
#     Girl(name='林心如'),
# ])
# session.add_all([
#     Boy2Girl(girl_id=1, boy_id=1),
#     Boy2Girl(girl_id=2, boy_id=1)
# ])


##### 要有girls = relationship('Girl', secondary='boy2girl', backref='boys')
# girl = Girl(name='張娜拉')
# girl.boys = [Boy(hostname='張鐵林'),Boy(hostname='費玉清')]
# session.add(girl)

# boy=Boy(hostname='蔡徐坤')
# boy.girls=[Girl(name='謝娜'),Girl(name='巧碧螺')]
# session.add(boy)
# session.commit()


# 基於對象的跨表查

# girl=session.query(Girl).filter_by(id=3).first()
# print(girl.boys)

#### 基於連表的跨表查詢

# 查詢蔡徐坤約過的所有妹子
'''
select girl.name from girl,boy,Boy2Girl where boy.id=Boy2Girl.boy_id and girl.id=Boy2Girl.girl_id where boy.name='蔡徐坤'

'''
# ret=session.query(Girl.name).filter(Boy.id==Boy2Girl.boy_id,Girl.id==Boy2Girl.girl_id,Boy.hostname=='蔡徐坤').all()

'''
select girl.name from girl inner join Boy2Girl on girl.id=Boy2Girl.girl_id inner join boy on boy.id=Boy2Girl.boy_id where boy.hostname='蔡徐坤'

'''
# ret=session.query(Girl.name).join(Boy2Girl).join(Boy).filter(Boy.hostname=='蔡徐坤').all()
ret=session.query(Girl.name).join(Boy2Girl).join(Boy).filter_by(hostname='蔡徐坤').all()
print(ret)


### 執行原生sql（用的最多的）
### django中orm如何執行原生sql
#
# cursor = session.execute('insert into users(name) values(:value)',params={"value":'xxx'})
# print(cursor.lastrowid)
# session.commit()

session.close()



res = session.squery(User.name.label('xx')).first()
res.xx  #label 相當於起別名

11.其它

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

# 關聯子查詢:correlate(Group)表示跟Group表做關聯，as_scalar相當於對該sql加括號，用於放在后面當子查詢
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`
"""

select id，name,
(select avr(score) from 成績表 where 成績表.sid=學生表.id) as x
from 學生表
subqry = session.query(func.count(成績表.scort).label("sc")).filter(學生表.id == 成績表.sid).correlate(學生表).as_scalar()
result = session.query(學生表.name, subqry)


session.close()

12.Flask-SQLAlchemy

1 Flask-SQLAlchemy
2 flask-migrate
    -python3 manage.py db init 初始化：只執行一次
    -python3 manage.py db migrate 等同於 makemigartions
    -python3 manage.py db upgrade 等同於migrate
    
3 看代碼
4 Flask-SQLAlchemy如何使用
	1 from flask_sqlalchemy import SQLAlchemy
	2 db = SQLAlchemy()
    3 db.init_app(app)
    4 以后在視圖函數中使用
    	-db.session 就是咱們講的session
        
5 flask-migrate的使用（表創建，字段修改）
	1 from flask_migrate import Migrate,MigrateCommand
    2 Migrate(app,db)
	3 manager.add_command('db', MigrateCommand)
6 直接使用
    -python3 manage.py db init 初始化：只執行一次，創建migrations文件夾
    -python3 manage.py db migrate 等同於 makemigartions
    -python3 manage.py db upgrade 等同於migrate