Flask之model以及緩存

Flask默認並沒有提供任何數據庫操作的API。

Flask中可以自己的選擇數據，用原生語句實現功能，也可以選擇ORM（SQLAlchemy，MongoEngine）
原生SQL缺點
代碼利用率低，條件復雜代碼語句越長，有很多相似語句
一些SQL是在業務邏輯中拼出來的，修改需要了解業務邏輯
直接寫SQL容易忽視SQL問題。

一、orm

將對對象的操作轉換為原生SQL

1、優點

易用性，可以有效減少重復SQL，性能損耗少設計靈活，可以輕松實現復雜查詢，移植性好

 Python中的orm是SQLAlchemy

針對於Flask的支持

pip install flask-sqlalchemy

2、連接數據庫

dialect+driver://username:password@host:port/database

dialect數據庫實現

driver數據庫的驅動

username

password

host

port

database

連接數據庫需要指定配置

app.config[‘SQLALCHEMY_DATABASE_URI’] = DB_URI
app.config[‘SQLALCHEMY_TRAKE_MODIFICATIONS’]=False

3、創建模型

class User(db.Model):
    __tablename__ = "UserModel"   # 指定表名，默認是類名

    id = db.Column(db.Integer, primary_key=True, autoincrement=True)

    u_name = db.Column(db.String(16), unique=True)

    u_des = db.Column(db.String(128), nullable=True)

（1）、字段類型

Integer            
SmallInteger        
BigInteger        
Float
Numeric
String
Text
Unicode
Unicode Text
Boolean
Date
Time
DateTime
Interval
LargeBinary

（2）、常見約束

primary_key
autoincrement
unique
index
nullable
default
ForeignKey()

（3）、數據操作

db.create_all()   創建數據庫

db.drop_all()    刪除數據庫

①、數據插入

數據插入是在事務中處理

db.session.add(object)
db.session.add_all(list[object])
db.session.commit()

@api.route('/adduser/')
def adduser():
    users = []
    for i in range(5):
        user = User()
        user.u_name = "小花%d" % random.randrange(10000)
        users.append(user)
    db.session.add_all(users)
    db.session.commit()
    return 'Add success'

②、數據刪除

db.session.delete(object)
db.session.commit()

修改和刪除基於查詢。

（4）、模型繼承

默認繼承並不會報錯，它會將多個模型的數據映射到一張表中，導致數據混亂，不能滿足基本使用
抽象的模型是不會在數據庫中產生映射的

class Animal(db.Model):
    __abstract__ = True
    id = db.Column(db.Integer, primary_key=True, autoincrement=True)
    a_name = db.Column(db.String(16))


class Dog(Animal):
    d_legs = db.Column(db.Integer, default=4)


class Cat(Animal):

    c_eat = db.Column(db.String(32), default='fish')

（5）、模型遷移

python manager.py db init   #初次遷移，生成migration包
python manager.py db migrate   # 創建遷移
python manager.py db upgrade   # 更新

（6）、數據查詢

①、查詢單個對象

first

get

get_or_404

@api.route('/getuser/<int:id>/')
def get_user(id):
    user = User.query.get(id)
    print(user)
    return 'GET success'

②、查詢結果集

all:比較特殊，返回列表

@api.route('/getusers/')
def get_users():
    users = User.query.all()
    for user in users:
        print(user.u_name)
    return 'get success'

filter：BaseQuery對象

運算符：

　　contains
　　startswith
　　endswith
　　in_
　　like
　　__gt__
　　__ge__
　　__lt__
　　__le__

條件：

- 類名.屬性名.魔術方法（臨界值）

@api.route('/getdog/')
def getdog():
    dogs = Dog.query.filter(Dog.id.__le__(5))
    for dog in dogs:
        print(dog.id, dog.a_name)
    return 'GET SUCCESS'

- 類名.屬性名 操作符運算符 臨界值

@api.route('/getdog/')
def getdog():
    dogs = Dog.query.filter(Dog.id > 5)
    for dog in dogs:
        print(dog.id, dog.a_name)
    return 'GET SUCCESS'

 

@api.route('/getdog/')
def getdog():
    dogs = Dog.query.filter(Dog.a_name.contains("2"))
    for dog in dogs:
        print(dog.id, dog.a_name)
    return 'GET SUCCESS'

offset和limit不區分順序，都是先執行offset

@api.route('/getdog/')
def getdog():
    dogs = Dog.query.offset(5).limit(4)
    for dog in dogs:
        print(dog.id, dog.a_name)
    return 'GET SUCCESS'

- order_by 調用必須在 offset和limit 之前

使用offset以及limit實現分頁

@api.route('/getdogs/')
def get_dogs():
    page = request.args.get("page", 1, type=int)
    per_page = request.args.get('per_page', 4, type=int)
    dogs = Dog.query.offset(per_page * (page - 1)).limit(per_page)
    return render_template('Dogs.html', dogs=dogs)

paginate實現分頁

@api.route('/getdogs/')
def get_dogs_with_page():
    # dogs = Dog.query.paginate().items
    pagination = Dog.query.paginate()
    per_page = request.args.get('per_page', 4, type=int)
    return render_template('Dogs.html', pagination=pagination, per_page=per_page)

<div class=pagination>
    {% for page in pagination.iter_pages(left_edge=5,left_current=5,right_current=5,right_edge=5) %}
        {% if page %}
            {% if page != pagination.page %}
                <a href="{{ url_for('api.get_dogs_with_page') }}?page={{ page }}&per_page={{ per_page }}">{{ page }}</a>
            {% else %}
                <strong>{{ page }}</strong>
            {% endif %}
        {% else %}
            <span class=ellipsis>…</span>
        {% endif %}
    {% endfor %}
</div>

filter_by

用在級聯數據上，條件語法精准，字段  =  值

@blue.route('/getcatsfilterby/')
def get_cats_filter_by():

    cats = Cat.query.filter_by(id = 5)

    return render_template('Cats.html', cats=cats)

（7）級聯數據

class Customer(db.Model):
    id = db.Column(db.Integer, primary_key=True, autoincrement=True)
    c_name = db.Column(db.String(16))
    addresses = db.relationship('Address', backref='customer', lazy=True)


class Address(db.Model):
    id = db.Column(db.Integer, primary_key=True, autoincrement=True)
    a_position = db.Column(db.String(128))
    a_customer_id = db.Column(db.Integer, db.ForeignKey(Customer.id))

 

 ①添加數據

@api.route('/getcustomer/')
def get():
    customer = Customer.query.order_by(desc('id')).first()

    return str(customer.id)


@api.route('/addaddress/')
def add_address():
    address = Address()
    address.a_position = '秀水街 %s' % random.randrange(10000)
    address.a_customer_id = Customer.query.order_by(desc('id')).first().id    # 注意此處使用id的倒敘，不能直接用‘-id’

    db.session.add(address)
    db.session.commit()
    return 'Address Add Success %s' % address.a_position

 

 ②查詢數據

根據地址找到對應的人

@api.route('/getcustomer/')
def get():
    a_id = request.args.get('a_id', type=int)
    address = Address.query.get(a_id)
    customer = Customer.query.get(address.a_customer_id)   #維護關系表中的外鍵存的是不維護關系表中的主鍵，
    return customer.c_name

 

 根據人找到對應的地址

@api.route('/getaddress/')
def get_address():
    c_id = request.args.get('c_id')
    customer = Customer.query.get(c_id)
   # addresses = Address.query.filter_by(a_customer_id=customer.id)
　　 addresses = customer.addresses
    return render_template('address.html', addresses=addresses)

<ul>
    {% for address in addresses %}
        <li>{{ address.a_position }}</li>
    {% endfor %}

</ul>

 

 ③邏輯運算

filter多個條件

@api.route('/getaddress/')
def get_address():
    addresses = Address.query.filter(Address.a_customer_id.__eq__(1)).filter(Address.a_position.endswith('4'))

    return render_template('address.html', addresses=addresses)

與  and   

filter(and_(條件),條件…)

@api.route('/getaddress/')
def get_address():
    addresses = Address.query.filter(and_(Address.a_customer_id.__eq__(1),Address.a_position.endswith('4')))

    return render_template('address.html', addresses=addresses)

Django中可以將字段直接寫在filter中，無需使用and_

 

或
or_
filter(or_(條件),條件…)

非
not_
filter(not_(條件),條件…)

@api.route('/getaddress/')
def get_address():
    addresses = Address.query.filter(not_(or_(Address.a_customer_id.__eq__(1), Address.a_position.endswith('4'))))

    return render_template('address.html', addresses=addresses)

 

二、緩存

pip install Flask-Caching

在ext.py中進行配置

from flask_caching import Cache
from flask_migrate import Migrate
from flask_session import Session
from flask_sqlalchemy import SQLAlchemy

db = SQLAlchemy()
migrate = Migrate()
cache = Cache(config={
    "CACHE_TYPE": "redis"    # 默認是連接本地，可以設置遠程。
})
# cache=Cache()   # 配置可以寫在settings中的Config類中

def init_ext(app):
    db.init_app(app)
    migrate.init_app(app, db)
    Session(app)
    cache.init_app(app)

在視圖中使用

@api.route('/getaddress/')
@cache.cached(timeout=60)
def get_address():
    addresses = Address.query.filter(not_(or_(Address.a_customer_id.__eq__(1), Address.a_position.endswith('4'))))
    print('數據庫中獲取')
    return render_template('address.html', addresses=addresses)