小知識：

1、子類繼承父類的三種方式

class Dog(Animal): #子類 派生類def __init__(self,name,breed, life_value,aggr):# Animal.__init__(self,name,breed, life_value,aggr)#讓子類執行父類的方法 就是父類名.方法名（參數），連self都得傳super().__init__(name,life_value,aggr) #super關鍵字 ，都不用傳self了，在新式類里的# super(Dog,self).__init__(name,life_value,aggr) #上面super是簡寫self.breed = breeddef bite(self,person): #狗的派生方法person.life_value -= self.aggrdef eat(self): #父類方法的重寫super().eat()print('dog is eating')

2、對象通過索引設置值的三種方式

方式一：重寫__setitem__方法

class Foo(object):def __setitem__(self, key, value):print(key,value)obj = Foo() obj["xxx"] = 123 #給對象賦值就會去執行__setitem__方法

方式二：繼承dict

class Foo(dict):passobj = Foo() obj["xxx"] = 123 print(obj)

方式三：繼承dict,重寫__init__方法的時候，記得要繼承父類的__init__方法

class Foo(dict):def __init__(self,val):# dict.__init__(self, val)#繼承父類方式一# super().__init__(val) #繼承父類方式二super(Foo,self).__init__(val)#繼承父類方式三 obj = Foo({"xxx":123}) print(obj)

總結：如果遇到obj["xxx"] = xx? ,　

- 重寫了__setitem__方法
- 繼承dict

3、測試__name__方法

示例：

app1中：import app2print('app1', __name__)app2中：print('app2', __name__)

現在app1是主程序，運行結果截圖

總結：如果是在自己的模塊中運行，__name__就是__main__，如果是從別的文件中導入進來的，就不是__name__了

一、設置配置文件的幾種方式

==========方式一：============app.config['SESSION_COOKIE_NAME'] = 'session_lvning' #這種方式要把所有的配置都放在一個文件夾里面，看起來會比較亂，所以選擇下面的方式 ==========方式二：============== app.config.from_pyfile('settings.py') #找到配置文件路徑，創建一個模塊，打開文件，并獲取所有的內容，再將配置文件中的所有值，都封裝到上一步創建的配置文件模板中

print(app.config.get("CCC")) =========方式三：對象的方式============
import os
os.environ['FLAKS-SETTINGS'] = 'settings.py'
app.config.from_envvar('FLAKS-SETTINGS')

===============方式四（推薦）：字符串的方式，方便操作，不用去改配置，直接改變字符串就行了 ==============
app.config.from_object('settings.DevConfig')

----------settings.DevConfig---------- from app import app
class BaseConfig(object):
NNN = 123 #注意是大寫
SESSION_COOKIE_NAME = "session_sss"

class TestConfig(BaseConfig):
DB = "127.0.0.1"

class DevConfig(BaseConfig):
DB = "52.5.7.5"

class ProConfig(BaseConfig):
DB = "55.4.22.4"

要想在視圖函數中獲取配置文件的值，都是通過app.config來拿。但是如果視圖函數和Flask創建的對象app不在一個模塊。就得

導入來拿。可以不用導入，。直接導入一個current_app，這個就是當前的app對象，用current_app.config就能查看到了當前app的所有的配置文件

from flask import Flask,current_app

?

@app.route('/index',methods=["GET","POST"]) def index():print(current_app.config) #當前的app的所有配置session["xx"] = "fdvbn"return "index"

?

二、藍圖（flask中多py文件拆分都要用到藍圖）

如果代碼非常多，要進行歸類。不同的功能放在不同的文件，吧相關的視圖函數也放進去。藍圖也就是對flask的目錄結構進行分配（應用于小，中型的程序），

小中型：

manage.py

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

__init__.py(只要一導入fcrm就會執行__init__.py文件)

from flask import Flask #導入accout 和order from fcrm.views import accout from fcrm.views import order app = Flask(__name__) print(app.root_path) #根目錄app.register_blueprint(accout.accout) #吧藍圖注冊到app里面，accout.accout是創建的藍圖對象 app.register_blueprint(order.order)

accout.py

from flask import Blueprint,render_template accout = Blueprint("accout",__name__)@accout.route('/accout') def xx():return "accout"@accout.route("/login") def login():return render_template("login.html")

order.py

from flask import Blueprint order = Blueprint("order",__name__)@order.route('/order') def register(): #注意視圖函數的名字不能和藍圖對象的名字一樣return "order

使用藍圖時需要注意的

大型：

?

三、數據庫連接池

flask中是沒有ORM的，如果在flask里面連接數據庫有兩種方式

一：pymysql 二：SQLAlchemy是python 操作數據庫的一個庫。能夠進行 orm 映射官方文檔 sqlchemySQLAlchemy“采用簡單的Python語言，為高效和高性能的數據庫訪問設計，實現了完整的企業級持久模型”。SQLAlchemy的理念是，SQL數據庫的量級和性能重要于對象集合；而對象集合的抽象又重要于表和行。

?鏈接池原理

- BDUtils數據庫鏈接池 - 模式一：基于threaing.local實現為每一個線程創建一個連接，關閉是偽關閉，當前線程可以重復- 模式二：連接池原理- 可以設置連接池中最大連接數 9- 默認啟動時，連接池中創建連接 5- 如果有三個線程來數據庫中獲取連接：- 如果三個同時來的，一人給一個鏈接- 如果一個一個來，有時間間隔，用一個鏈接就可以為三個線程提供服務- 說不準有可能:1個鏈接就可以為三個線程提供服務有可能:2個鏈接就可以為三個線程提供服務有可能:3個鏈接就可以為三個線程提供服務PS、：maxshared在使用pymysql中均無用。鏈接數據庫的模塊：只有threadsafety>1的時候才有用

?

那么我們用pymysql來做。

為什么要使用數據庫連接池呢？不用連接池有什么不好的地方呢？

方式一、每次操作都要鏈接數據庫，鏈接次數過多

#!usr/bin/env python # -*- coding:utf-8 -*- import pymysql from flask import Flaskapp = Flask(__name__)# 方式一：這種方式每次請求，反復創建數據庫鏈接，多次鏈接數據庫會非常耗時 # 解決辦法：放在全局，單例模式 @app.route('/index') def index():# 鏈接數據庫conn = pymysql.connect(host="127.0.0.1",port=3306,user='root',password='123', database='pooldb',charset='utf8')cursor = conn.cursor()cursor.execute("select * from td where id=%s", [5, ])result = cursor.fetchall() # 獲取數據cursor.close()conn.close() # 關閉鏈接print(result)return "執行成功"if __name__ == '__main__':app.run(debug=True)

方式二、不支持并發

#!usr/bin/env python # -*- coding:utf-8 -*- import pymysql from flask import Flask from threading import RLockapp = Flask(__name__) CONN = pymysql.connect(host="127.0.0.1",port=3306,user='root',password='123', database='pooldb',charset='utf8') # 方式二：放在全局，如果是單線程，這樣就可以，但是如果是多線程，就得加把鎖。這樣就成串行的了 # 不支持并發，也不好。所有我們選擇用數據庫連接池 @app.route('/index') def index():with RLock:cursor = CONN.cursor()cursor.execute("select * from td where id=%s", [5, ])result = cursor.fetchall() # 獲取數據cursor.close()print(result)return "執行成功" if __name__ == '__main__':app.run(debug=True)

?

方式三：由于上面兩種方案都不完美，所以得把方式一和方式二聯合一下（既讓減少鏈接次數，也能支持并發）所有了方式三，需要

導入一個DButils模塊

基于DButils實現的數據庫連接池有兩種模式：

模式一：為每一個線程創建一個鏈接（是基于本地線程來實現的。thread.local），每個線程獨立使用自己的數據庫鏈接，該線程關閉不是真正的關閉，本線程再次調用時，還是使用的最開始創建的鏈接，直到線程終止，數據庫鏈接才關閉

注：?模式一：如果線程比較多還是會創建很多連接，模式二更常用?

#!usr/bin/env python # -*- coding:utf-8 -*- from flask import Flask app = Flask(__name__) from DBUtils.PersistentDB import PersistentDB import pymysql POOL = PersistentDB(creator=pymysql, # 使用鏈接數據庫的模塊maxusage=None, # 一個鏈接最多被重復使用的次數，None表示無限制setsession=[], # 開始會話前執行的命令列表。如：["set datestyle to ...", "set time zone ..."]ping=0,# ping MySQL服務端，檢查是否服務可用。# 如：0 = None = never, 1 = default = whenever it is requested, 2 = when a cursor is created, 4 = when a query is executed, 7 = alwayscloseable=False,# 如果為False時， conn.close() 實際上被忽略，供下次使用，再線程關閉時，才會自動關閉鏈接。如果為True時， conn.close()則關閉鏈接，那么再次調用pool.connection時就會報錯，因為已經真的關閉了連接（pool.steady_connection()可以獲取一個新的鏈接）threadlocal=None, # 本線程獨享值得對象，用于保存鏈接對象，如果鏈接對象被重置host='127.0.0.1',port=3306,user='root',password='123',database='pooldb',charset='utf8' )@app.route('/func') def func(): 　　conn = POOL.connection()
　　cursor = conn.cursor()
　　cursor.execute('select * from tb1')
　　result = cursor.fetchall()
　　cursor.close()
　　conn.close() # 不是真的關閉，而是假的關閉。 conn = pymysql.connect() conn.close()

　　conn = POOL.connection()
　　cursor = conn.cursor()
　　cursor.execute('select * from tb1')
　　result = cursor.fetchall()
　　cursor.close()
　　conn.close() if __name__ == '__main__': app.run(debug=True)

模式二：創建一個鏈接池，為所有線程提供連接，使用時來進行獲取，使用完畢后在放回到連接池。

　　　　PS：假設最大鏈接數有10個，其實也就是一個列表，當你pop一個，人家會在append一個，鏈接池的所有的鏈接都是按照排隊的這樣的方式來鏈接的。

　　　　　鏈接池里所有的鏈接都能重復使用，共享的，?即實現了并發，又防止了鏈接次數太多

import time import pymysql import threading from DBUtils.PooledDB import PooledDB, SharedDBConnection POOL = PooledDB(creator=pymysql, # 使用鏈接數據庫的模塊maxconnections=6, # 連接池允許的最大連接數，0和None表示不限制連接數mincached=2, # 初始化時，鏈接池中至少創建的空閑的鏈接，0表示不創建maxcached=5, # 鏈接池中最多閑置的鏈接，0和None不限制maxshared=3, # 鏈接池中最多共享的鏈接數量，0和None表示全部共享。PS: 無用，因為pymysql和MySQLdb等模塊的 threadsafety都為1，所有值無論設置為多少，_maxcached永遠為0，所以永遠是所有鏈接都共享。blocking=True, # 連接池中如果沒有可用連接后，是否阻塞等待。True，等待；False，不等待然后報錯maxusage=None, # 一個鏈接最多被重復使用的次數，None表示無限制setsession=[], # 開始會話前執行的命令列表。如：["set datestyle to ...", "set time zone ..."]ping=0,# ping MySQL服務端，檢查是否服務可用。# 如：0 = None = never, 1 = default = whenever it is requested, 2 = when a cursor is created, 4 = when a query is executed, 7 = alwayshost='127.0.0.1',port=3306,user='root',password='123',database='pooldb',charset='utf8' )def func():# 檢測當前正在運行連接數的是否小于最大鏈接數，如果不小于則：等待或報raise TooManyConnections異常# 否則# 則優先去初始化時創建的鏈接中獲取鏈接 SteadyDBConnection。# 然后將SteadyDBConnection對象封裝到PooledDedicatedDBConnection中并返回。# 如果最開始創建的鏈接沒有鏈接，則去創建一個SteadyDBConnection對象，再封裝到PooledDedicatedDBConnection中并返回。# 一旦關閉鏈接后，連接就返回到連接池讓后續線程繼續使用。# PooledDedicatedDBConnectionconn = POOL.connection()# print(th, '鏈接被拿走了', conn1._con)# print(th, '池子里目前有', pool._idle_cache, '\r\n')cursor = conn.cursor()cursor.execute('select * from tb1')result = cursor.fetchall()conn.close()conn = POOL.connection()# print(th, '鏈接被拿走了', conn1._con)# print(th, '池子里目前有', pool._idle_cache, '\r\n')cursor = conn.cursor()cursor.execute('select * from tb1')result = cursor.fetchall()conn.close()func()

五、本地線程：保證每個線程都只有自己的一份數據，在操作時不會影響別人的，即使是多線程，自己的值也是互相隔離的

沒用線程之前

import threading
import time class Foo(object):def __init__(self):self.name = None local_values = Foo()def func(num):time.sleep(2)local_values.name = numprint(local_values.name,threading.current_thread().name)for i in range(5):th = threading.Thread(target=func, args=(i,), name='線程%s' % i)th.start()

?

打印結果：

1 線程1 0 線程0 2 線程2 3 線程3 4 線程4

用了本地線程之后

import threading import time # 本地線程對象 local_values = threading.local() def func(num):"""# 第一個線程進來，本地線程對象會為他創建一個# 第二個線程進來，本地線程對象會為他創建一個{線程1的唯一標識：{name:1},線程2的唯一標識：{name:2},}:param num::return:"""local_values.name = num # 4# 線程停下來了time.sleep(2)# 第二個線程： local_values.name，去local_values中根據自己的唯一標識作為key，獲取value中name對應的值print(local_values.name, threading.current_thread().name)for i in range(5):th = threading.Thread(target=func, args=(i,), name='線程%s' % i)th.start()

打印結果：

1 線程1 2 線程2 0 線程0 4 線程4 3 線程3

六、上下文管理

a、類似于本地線程創建Local類：{線程或協程唯一標識: { 'stack':[request],'xxx':[session,] },線程或協程唯一標識: { 'stack':[] },線程或協程唯一標識: { 'stack':[] },線程或協程唯一標識: { 'stack':[] },}b、上下文管理的本質每一個線程都會創建一個上面那樣的結構，當請求進來之后，將請求相關數據添加到列表里面[request，]，以后如果使用時，就去讀取列表中的數據，請求完成之后，將request從列表中移除c、關系local = 小華={線程或協程唯一標識: { 'stack':[] },線程或協程唯一標識: { 'stack':[] },線程或協程唯一標識: { 'stack':[] },線程或協程唯一標識: { 'stack':[] },}stack = 強哥 = {pushpoptop}存取東西時都要基于強哥來做d、最近看過一些flask源碼，flask還是django有些區別- Flask和Django區別？- 請求相關數據傳遞的方式- django：是通過傳request參數實現的- Flask：基于local對象和，localstark對象來完成的當請求剛進來的時候就給放進來了，完了top取值就行了,取完之后pop走就行了問題：多個請求過來會不會混淆-答： 不會，因為，不僅是線程的，還是協程，每一個協程都是有唯一標識的：from greenlent import getcurrentt as get_ident #這個就是來獲取唯一標識的

?

flask的request和session設置方式比較新穎，如果沒有這種方式，那么就只能通過參數的傳遞。

flask是如何做的呢？

- 本地線程：是Flask自己創建的一個線程（猜想：內部是不是基于本地線程做的？）vals = threading.local()def task(arg):vals.name = num- 每個線程進來都是打印的自己的，只有自己的才能修改，- 通過他就能保證每一個線程里面有一個數據庫鏈接，通過他就能創建出數據庫鏈接池的第一種模式- 上下文原理- 類似于本地線程- 猜想：內部是不是基于本地線程做的？不是，是一個特殊的字典

?

#!/usr/bin/env python # -*- coding:utf-8 -*- from functools import partial from flask.globals import LocalStack, LocalProxyls = LocalStack()class RequestContext(object):def __init__(self, environ):self.request = environdef _lookup_req_object(name):top = ls.topif top is None:raise RuntimeError(ls)return getattr(top, name)session = LocalProxy(partial(_lookup_req_object, 'request'))ls.push(RequestContext('c1')) # 當請求進來時，放入 print(session) # 視圖函數使用 print(session) # 視圖函數使用 ls.pop() # 請求結束popls.push(RequestContext('c2')) print(session)ls.push(RequestContext('c3')) print(session)

3. Flask內部實現

#!/usr/bin/env python # -*- coding:utf-8 -*-from greenlet import getcurrent as get_identdef release_local(local):local.__release_local__()class Local(object):__slots__ = ('__storage__', '__ident_func__')def __init__(self):# self.__storage__ = {}# self.__ident_func__ = get_identobject.__setattr__(self, '__storage__', {})object.__setattr__(self, '__ident_func__', get_ident)def __release_local__(self):self.__storage__.pop(self.__ident_func__(), None)def __getattr__(self, name):try:return self.__storage__[self.__ident_func__()][name]except KeyError:raise AttributeError(name)def __setattr__(self, name, value):ident = self.__ident_func__()storage = self.__storage__try:storage[ident][name] = valueexcept KeyError:storage[ident] = {name: value}def __delattr__(self, name):try:del self.__storage__[self.__ident_func__()][name]except KeyError:raise AttributeError(name)class LocalStack(object):def __init__(self):self._local = Local()def __release_local__(self):self._local.__release_local__()def push(self, obj):"""Pushes a new item to the stack"""rv = getattr(self._local, 'stack', None)if rv is None:self._local.stack = rv = []rv.append(obj)return rvdef pop(self):"""Removes the topmost item from the stack, will return theold value or `None` if the stack was already empty."""stack = getattr(self._local, 'stack', None)if stack is None:return Noneelif len(stack) == 1:release_local(self._local)return stack[-1]else:return stack.pop()@propertydef top(self):"""The topmost item on the stack. If the stack is empty,`None` is returned."""try:return self._local.stack[-1]except (AttributeError, IndexError):return Nonestc = LocalStack()stc.push(123) v = stc.pop()print(v)

?

轉載于:https://www.cnblogs.com/morgana/p/8493191.html

總結

以上是生活随笔為你收集整理的基于DBUtils实现数据库连接池的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。