共計 11272 個字符，預計需要花費 29 分鐘才能閱讀完成。
這篇文章給大家介紹 Dataset 中 SQLAlchemy 如何使用，內容非常詳細，感興趣的小伙伴們可以參考借鑒，希望對大家能有所幫助。
Dataset 對于操作 JSON、CSV 文件、NoSQL 非常好用。
import dataset連接 MySQL 數據庫：
db = dataset.connect(mysql://username:password@10.10.10.10/ctf?charset=utf8)用戶名:username，密碼:password，數據庫地址(地址 + 端口):10.10.10.10，database 名: ctf
連接 SQLite 數據庫：
db = dataset.connect(sqlite:///ctf.db)連接 PostgreSQL 數據庫：
db = dataset.connect(postgresql://scott:tiger@localhost:5432/mydatabase)一定要注意指定字符編碼
table = db[city] #（選擇 city 表） user = table(name) # 找出表中 name 列屬性所有數據 res = db.query(select name from table limit 10) # 如果不需要查看全部數據的話 *** 用 limit，因為全部數據的載入非常非常耗時間 for x in res: print x[name] # 選 name 字段的數據 table.insert(dict(name= John Doe , age=37)) table.insert(dict(name= Jane Doe , age=34, gender= female)) john = table.find_one(name= John Doe)在數據庫中查找是否有同時滿足多個條件的數據：table.find_one(屬性 1 = 屬性值 1, 屬性 2 = 屬性值 2, hellip;)
注：find_one 速度很慢
插入數據
dataset 會根據輸入自動創建表和字段名
table = db[user] # 或者 table = db.get_table(user) table.insert(dict(name= John Doe , age=46, country= China)) table.insert(dict(name= Jane Doe , age=37, country= France , gender= female)) # 主鍵 id 自動生成更新數據
table.update(dict(name= John Doe , age=47), [name]) # 第二個參數相當于 sql update 語句中的 where，用來過濾出需要更新的記錄事務操作
事務操作可以簡單的使用上下文管理器來實現, 出現異常，將會回滾
with dataset.connect() as tx: tx[ user].insert(dict(name= John Doe , age=46, country= China)) # 相當于： db = dataset.connect() db.begin() try: db[user].insert(dict(name= John Doe , age=46, country= China)) db.commit() except: db.rollback() # 也可以嵌套使用: db = dataset.connect() with db as tx1: tx1[ user].insert(dict(name= John Doe , age=46, country= China)) with db as tx2: tx2[user].insert(dict(name= Jane Doe , age=37, country= France , gender= female))從表獲取數據
users = db[user].all() for user in db[ user]: # print(user[ age]) # chinese_users = user.find(country= China) john = user.find_one(name= John Doe)獲取非重復數據
db[user].distinct(country)刪除記錄
table.delete(place= Berlin)執行 SQL 語句
result = db.query(SELECT country, COUNT(*) c FROM user GROUP BY country ) for row in result: print(row[ country], row[c])導出數據
result = db[users].all() dataset.freeze(result, format= json , filename= users.json)JSON
JSON(JavaScript Object Notation) 是一種輕量級的數據交換格式，非常易于人閱讀和編寫。
import jsonjson.dumps 將 Python 對象編碼成 JSON 字符串
json.loads 將已編碼的 JSON 字符串解碼為 Python 對象
MySQL 數據庫：
分類表 -categories，包括類別 web,reversing，crypto(加解密),mic 等
題目表 -tasks, 包括題目 id，題目名，flag, 分值，文件 地址，題目等級，題目詳細描述
flag 表 -flag, 包括題目 id, 用戶 id, 得分，時間戳
用戶表 -users，包括用戶 id, 用戶名，密碼
題目分類表 -cat_task, 包括題目 id, 題目類別 id
flag 表中每條數據由于是有題目 ID task_id 和用戶 ID user_id 來共同確認的，所以采用復合主鍵：primary key (task_id,user_id)
聯合主鍵和復合主鍵的區別
python 裝飾器
Decorator 通過返回包裝對象實現間接調用, 以此插入額外邏輯
https://www.zhihu.com/question/26930016
wraps 本身也是一個裝飾器，它能把原函數的元信息拷貝到裝飾器函數中，這使得裝飾器函數也有和原函數一樣的元信息了
from functools import wraps def logged(func): @wraps(func) def with_logging(*args,**kwargs): print func.__name__ + was called return func(*args,**kwargs) return with_logging @logged def f(x): does some math return x + x * x print f.__name__ # prints f print f.__doc__ # prints does some mathweb 框架采用 flask
from flask import Flask引入 Flask 類，Flask 類實現了一個 WSGI(Web Server Gateway Interface)應用
app = Flask(__name__)app 是 Flask 的實例，它接收包或者模塊的名字作為參數，但一般都是傳遞__name__
@app.route(/) def hello_world(): return Hello World!使用 app.route 裝飾器會將 URL 和執行的視圖函數的關系保存到 app.url_map 屬性上。處理 URL 和視圖函數的關系的程序就是路由，這里的視圖函數就是 hello_world
if __name__ == __main__ : app.run(host= 0.0.0.0 ,port=9000)使用這個判斷可以保證當其他文件引用這個文件的時候 (例如 from hello import app) 不會執行這個判斷內的代碼，也就是不會執行 app.run 函數。
執行 app.run 就可以啟動服務了。默認 Flask 只監聽虛擬機的本地 127.0.0.1 這個地址，端口為 5000。而我們對虛擬機做的端口轉發端口是 9000，所以需要制定 host 和 port 參數，0.0.0.0 表示監聽所有地址，這樣就可以在本機訪問了。
服務器啟動后，會調用 werkzeug.serving.run_simple 進入輪詢，默認使用單進程單線程的 werkzeug.serving.BaseWSGIServer 處理請求，實際上還是使用標準庫 BaseHTTPServer.HTTPServer，通過 select.select 做 0.5 秒的 while TRUE 的事件輪詢。當我們訪問 http://127.0.0.1:9000/, 通過 app.url_map 找到注冊的 / 這個 URL 模式, 就找到了對應的 hello_world 函數執行，返回 hello world!, 狀態碼為 200。如果訪問一個不存在的路徑，如訪問 http://127.0.0.1:9000/a,Flask 找不到對應的模式，就會向瀏覽器返回 Not Found，狀態碼為 404
flask 中 jsonify 的作用
jsonify 的作用實際上就是將我們傳入的 json 形式數據序列化成為 json 字符串，作為響應的 body，并且設置響應的 Content-Type 為 application/json，構造出響應返回至客戶端
效果等于 json.dumps
jsonify 的 Content-Type 字段值為 application/json
json.dumps 的 Content-Type 字段值為 text/html
修改 flask 中靜態文件夾
修改的 flask 默認的 static 文件夾只需要在創建 Flask 實例的時候, 把 static_folder 和 static_url_path 參數設置為空字符串即可。
app = Flask(__name__, static_folder=”, static_url_path=”)
訪問的時候用 url_for 函數，res 文件夾和 static 文件夾同一級：
url_for(lsquo;static rsquo;, filename= rsquo;res/favicon.ico rsquo;)
werkzeug
werkzeug 是一個 WSGI 工具包，可以作為一個 Web 框架的底層庫。它封裝好了很多 Web 框架的東西，例如 Request，Response 等等。Flask 框架就是一 Werkzeug 為基礎開發的
generate_password_hash(password)
將用戶輸入的明文密碼加密成密文進行存儲
密碼加鹽哈希函數。用來將明文密碼加密，返回加密后的密文，用來進行用戶注冊
函數定義：
werkzeug.security.generate_password_hash(password, method= pbkdf2:sha1 , salt_length=8)
密文格式：method$salt$hash
password: 明文密碼
method: 哈希的方式(需要是 hashlib 庫支持的)，格式為
pbpdf2: method [:iterations]。參數說明：
method：哈希的方式，一般為 SHA1，
iterations：(可選參數)迭代次數，默認為 1000。
slat_length: 鹽值的長度，默認為 8
check_password_hash(hash,password)
驗證經過 generate_password_hash 哈希的密碼，將明文和密文進行比較, 查看是否一致，用來驗證用戶登錄
函數定義：
werkzeug.security.check_password_hash(pwhash, password)
pwhash: generate_password_hash 生成的哈希字符串
password: 需要驗證的明文密碼
flask 中的 session
rom flask import session user = db[users].find_one(username=username) session[user_id] = user[id]由于使用了 session，所以需要設置一個 secret_key 用來做一些模塊的 hash
Flask Web Development 中的內容:
SECRET_KEY 配置變量是通用密鑰, 可在 Flask 和多個第三方擴展中使用。如其名所示, 加密的強度取決于變量值的機密度。不同的程序要使用不同的密鑰, 而且要保證其他人不知道你所用的字符串。
SECRET_KEY 的作用主要是提供一個值做各種 HASH, 是在其加密過程中作為算法的一個參數(salt 或其他)。所以這個值的復雜度也就影響到了數據傳輸和存儲時的復雜度。
flask 變量規則
要給 URL 添加變量部分，你可以把這些特殊的字段標記為 variable_name， 這個部分將會作為命名參數傳遞到你的函數。規則可以用 converter:variable_name 指定一個可選的轉換器
@route(/hello/ name) def index(name): return b Hello {{name}} /b !數據庫查詢
對 dataset 的數據查詢，使用冒號來為變量傳參。
select f.task_id from flags f where f.user_id = :user_id”rsquo;,user_id=session[lsquo;user_id rsquo;])
模板渲染
使用 render_template 方法來渲染模板。將模板名和你想作為關鍵字的參數傳入模板的變量
MySQL
IFNULL(expr1,expr2)
如果 expr1 不是 NULL，IFNULL()返回 expr1，否則它返回 expr2。
IFNULL()返回一個數字或字符串值，取決于它被使用的上下文環境。
max 函數是用來找出記錄集中 *** 值的記錄
對于 left join，不管 on 后面跟什么條件，左表的數據全部查出來，因此要想過濾需把條件放到 where 后面
對于 inner join，滿足 on 后面的條件表的數據才能查出，可以起到過濾作用。也可以把條件放到 where 后面
在使用 left jion 時，on 和 where 條件的區別如下：
on 條件是在生成臨時表時使用的條件，它不管 on 中的條件是否為真，都會返回左邊表中的記錄。
where 條件是在臨時表生成好后，再對臨時表進行過濾的條件。這時已經沒有 left join 的含義 (必須返回左邊表的記錄) 了，條件不為真的就全部過濾掉。
order by 的用法
使用 order by，一般是用來，依照查詢結果的某一列 (或多列) 屬性，進行排序(升序：ASC; 降序：DESC; 默認為升序)。
當排序列含空值時：
ASC：排序列為空值的元組 *** 顯示。
DESC：排序列為空值的元組 *** 顯示。
可以把 null 值看做無窮大
select * from s order by sno desc, sage asc
group by 的用法
group by 按照查詢結果集中的某一列(或多列)，進行分組，值相等的為一組
1、細化集函數 (count，sum，avg，max，min) 的作用對象：
未對查詢結果分組，集函數將作用于整個查詢結果。
對查詢結果分組后，集函數將分別作用于每個組。
SELECT cno,count(sno) from sc group by cno
2、GROUP BY 子句的作用對象是查詢的中間結果表
分組方法：按指定的一列或多列值分組，值相等的為一組。
使用 GROUP BY 子句后，SELECT 子句的列名列表中只能出現分組屬性 (比如：sno) 和集函數(比如：count())
select sno,count(cno) from sc group by sno
3、多個列屬性進行分組
select cno,grade,count(cno) from sc group by cno,grade
4、使用 HAVING 短語篩選最終輸出結果
只有滿足 HAVING 短語指定條件的組才輸出。
HAVING 短語與 WHERE 子句的區別：作用對象不同。
1、WHERE 子句作用于基表或視圖，從中選擇滿足條件的元組。
2、HAVING 短語作用于組，從中選擇滿足條件的組
select sno from sc group by sno having count(cno) 3
select sno,count(cno) from sc where grade 60 group by sno having count(cno) 3
MySQL 的左連接、右連接、等值連接
1. 左連接(left join)
select m.columnname hellip; hellip;,n.* columnname hellip;.. from left_table m left join right_table n on m.columnname_join=n.columnname_join and n.columnname=xxx where m.columnname=xxx hellip;..ON 是連接條件，用于把 2 表中等值的記錄連接在一起，但是不影響記錄集的數量。若是表 left_table 中的某記錄，無法在表 right_table 找到對應的記錄，則此記錄依然顯示在記錄集中，只是表 right_table 需要在查詢顯示的列的值用 NULL 替代;
ON 連接條件中表 n.columnname=xxx 用于控制 right_table 表是否有符合要求的列值還是用 NULL 替換的方式顯示在查詢列中，不影響記錄集的數量;
WHERE 字句控制記錄是否符合查詢要求，不符合則過濾掉
2. 右連接(right join)
select m.columnname hellip; hellip;,n.* columnname hellip;.. from left_table m right join right_table n on m. columnname_join=n. columnname_join and m. columnname=xxx where n.columnname=xxx hellip;..3. 等值連接
select m.columnname hellip; hellip;,n.* columnname hellip;.. from left_table m [inner] join right_table n on m. columnname_join=n. columnname_join where m.columnname=xxx hellip;.. and n.columnname=xxx hellip;.或者
select m.columnname hellip; hellip;,n.* columnname hellip;.. from left_table m , right_table n where m. columnname_join=n. columnname_join and m.columnname=xxx hellip;.. and n.columnname=xxx hellip;.ON 是連接條件，不再與左連接或右連接的功效一樣，除了作為 2 表記錄匹配的條件外，還會起到過濾記錄的作用，若 left_table 中記錄無法在 right_table 中找到對應的記錄，則會被過濾掉;
WHERE 字句，不管是涉及表 left_table、表 right_table 上的限制條件，還是涉及 2 表連接的條件，都會對記錄集起到過濾作用，把不符合要求的記錄刷選掉;
jinja2 獲取循環索引
jinja2 獲取循環 {% for i in n %} 的索引使用 loop.index
{% for i in names %} tr td {{ loop.index }} /td // 當前是第 x 條 td {{ i.name }} /td /tr {% endfor %}flask 重定向和錯誤
可以用 redirect()函數把用戶重定向到其它地方。放棄請求并返回錯誤代碼，用 abort()函數。
from flask import abort, redirect, url_for @app.route(/) def index(): return redirect(url_for( login)) @app.route(/login) def login(): abort(401) this_is_never_executed()默認情況下，錯誤代碼會顯示一個黑白的錯誤頁面。如果你要定制錯誤頁面，可以使用 errorhandler()
裝飾器:
from flask import render_template @app.errorhandler(404) def page_not_found(error): return render_template(page_not_found.html), 404注意 render_template()調用之后的 404。這告訴 Flask，該頁的錯誤代碼是 404 ，即沒有找到。默認為 200，也就是一切正常。
flask CSRF 防護機制
@app.before_request def csrf_protect(): if request.method == POST : token = session.pop( _csrf_token , None) if not token or token != request.form.get(_csrf_token): abort(403) def some_random_string(): return hashlib.sha256(os.urandom(16).hexdigest()) def generate_csrf_token(): if _csrf_token not in session: session[ _csrf_token] = some_random_string() return session[ _csrf_token]在 flask 的全局變量里面注冊 上面那個生成隨機 token 的函數
app.jinja_env.globals[lsquo;csrf_token rsquo;] = generate_csrf_token
在網頁的模板是這么引入的
form method=post action= input name=_csrf_token type=hidden value= {{ csrf_token() }}flask 上下文處理器
Flask 上下文處理器自動向模板的上下文中插入新變量。上下文處理器在模板渲染之前運行，并且可以在模板上下文中插入新值。上下文處理器是一個返回字典的函數，這個字典的鍵值最終將傳入應用中所有模板的上下文:
@app.context_processor def inject_user(): return dict(user=g.user)上面的上下文處理器使得模板可以使用一個名為 user 值為 g.user 的變量。不過這個例子不是很有意思，因為 g 在模板中本來就是可用的，但它解釋了上下文處理器是如何工作的。
變量不僅限于值，上下文處理器也可以使某個函數在模板中可用(由于 Python 允許傳遞函數):
@app.context_processor def utility_processor(): def format_price(amount, currency=u euro;): return u {0:.2f}{1}.format(amount, currency) return dict(format_price=format_price)上面的上下文處理器使得 format_price 函數在所有模板中可用:
{{format_price(0.33) }}
日志記錄
handler = logging.FileHandler(lsquo;flask.log rsquo;, encoding= rsquo;UTF-8 prime;)
1、請求之前設置 requestId 并記錄日志
每個 URL 請求之前，定義 requestId 并綁定到 g
@app.before_request def before_request(): g.requestId = gen_requestId() logger.info(Start Once Access, and this requestId is %s % g.requestId)2、請求之后添加響應頭與記錄日志
每次返回數據中，帶上響應頭，包含 API 版本和本次請求的 requestId，以及允許所有域跨域訪問 API, 記錄訪問日志
@app.after_request def add_header(response): response.headers[X-SaintIC-Media-Type] = saintic.v1 response.headers[X-SaintIC-Request-Id] = g.requestId response.headers[Access-Control-Allow-Origin] = * logger.info(json.dumps({ AccessLog : { status_code : response.status_code, method : request.method, ip : request.headers.get( X-Real-Ip , request.remote_addr), url : request.url, referer : request.headers.get(Referer), agent : request.headers.get(User-Agent), requestId : str(g.requestId), } } )) return responsebasicConfig 方法可以滿足你在絕大多數場景下的使用需求，但是 basicConfig 有一個很大的缺點。調用 basicConfig 其實是給 root logger 添加了一個 handler(FileHandler)，這樣當你的程序和別的使用了 logging 的第三方模塊一起工作時，會影響第三方模塊的 logger 行為。這是由 logger 的繼承特性決定的
logging.basicConfig(level=logging.DEBUG, format= %(asctime)s %(levelname)s %(message)s , datefmt= %a, %d %b %Y %H:%M:%S , filename= logs/pro.log , filemode= w ) logging.debug(dddddddddd)MySQL 字符編碼
除了設置數據庫的之外，由于 dataset 默認創建數據庫和表的字符集不是 utf8, 所以需要自己設置，否則會中文亂碼，所以需要修改表的字符集
my.cnf [client] default-character-set=utf8 [mysqld] character-set-server=utf8 collation-server=utf8_general_ci default-storage-engine=INNODB表的字符集
show create table tasks; alter table tasks convert to character set utf8;關于 Dataset 中 SQLAlchemy 如何使用就分享到這里了，希望以上內容可以對大家有一定的幫助，可以學到更多知識。如果覺得文章不錯，可以把它分享出去讓更多的人看到。
