Sanic 基礎應用 -2
Sanic 基礎 - 2
1. 異常的處理
NotFound: 找不到合適的路由請求。ServerError: 服務器內部出現問題時調用。一般發生在用戶代碼出現錯誤的狀況。
from sanic import Sanic
from sanic.exceptions import NotFound
from sanic.exceptions import ServerError
from sanic.response import text
app = Sanic(__name__)
@app.exception(NotFound)
def ignore_404s(request, exception):
# 捕獲頁面找不到異常 404
return text("Yep, I totally found the page: {}".format(request.url))
@app.route('/error')
def i_am_ready_to_die(request):
raise ServerError("拋出一個異常!!", status_code=500)
if __name__ == '__main__':
app.run()
2. 中間件和監聽器
中間件:
- 這裏有兩種不一樣類型的中間件：請求request和響應response。 都是使用
@app.middleware裝飾器進行聲明的，利用'request'或'response'字符串來表示其參數類型。
from sanic import Sanic
from sanic.exceptions import NotFound
from sanic.exceptions import ServerError
from sanic.response import text
app = Sanic(__name__)
@app.middleware('request')
async def print_on_request(request):
print("當服務器接收到請求時，我打印")
return text('I halted the request') # 修改響應內容
@app.middleware('response')
async def add_headers(request, response):
response.headers["Server"] = "Fake-Server"
@app.middleware('response')
async def add_headers(request, response):
print("當服務器返回響應時打印")
# 添加響應頭
response.headers["x-xss-protection"] = "1; mode=block" # 將添加HTTP頭以防止跨站點腳本（XSS）攻擊。
# response.headers["Server"] = "Fake-Server"
return text('I halted the response') # 修改響應內容 就不會設置響應頭了
@app.route('/')
async def index(request):
return text('OK')
if __name__ == '__main__':
app.run()
監聽器:
from sanic import Sanic
from sanic.response import text
app = Sanic(__name__)
@app.listener('before_server_start')
async def setup_db(app, loop):
# app.db = await db_setup()
print("服務器開啓時")
@app.listener('after_server_start')
async def notify_server_started(app, loop):
print("服務器成功啓動時")
@app.listener('before_server_stop')
async def notify_server_stopping(app, loop):
print('服務器關閉以前')
@app.listener('after_server_stop')
async def close_db(app, loop):
print('服務器關閉以後')
await app.db.close()
async def notify_server_started_after_five_seconds():
await asyncio.sleep(5)
print('Server successfully started!')
app.add_task(notify_server_started_after_five_seconds())
@app.route('/')
async def index(request):
return text('OK')
if __name__ == '__main__':
app.run()
3. 藍圖
3.1 實現並註冊
藍圖python
from sanic.response import json from sanic import Blueprint, Sanic # bp = Blueprint('my_blueprint') bp = Blueprint(__name__) # 名字隨意 @bp.route('/') async def bp_root(request): return json({'my': 'blueprint'})註冊mysql
from sanic import Sanic from my_blueprint import bp app = Sanic(__name__) app.blueprint(bp) app.run(host='0.0.0.0', port=8000, debug=True)
可全局註冊中間件web
@bp.middleware async def print_on_request(request): print("I am a spy") @bp.middleware('request') async def halt_request(request): return text('I halted the request') @bp.middleware('response') async def halt_response(request, response): return text('I halted the response')全局處理異常sql
@bp.exception(NotFound) def ignore_404s(request, exception): return text("Yep, I totally found the page: {}".format(request.url))靜態文件json
bp.static('/folder/to/serve', '/web/path')路由反響解析api
from sanic.response import text, redirect from sanic import Blueprint, Sanic app = Sanic() blueprint_v1 = Blueprint('v1', url_prefix='/v1') app.blueprint(blueprint_v1) @blueprint_v1.route('/post/<arg>', name='post_handler') async def post_handler(request, arg): return text('Post {} in Blueprint V1'.format(arg)) print(app.url_for('v1.post_handler', arg=123)) # /v1/post/123監聽瀏覽器
@bp.listener('before_server_start') async def setup_connection(app, loop): # global database # database = mysql.connect(host='127.0.0.1'...) print('開始') @bp.listener('after_server_stop') async def close_connection(app, loop): print('結束') # await database.close()
3.2 API版本控制
lueprints對於API版本控制很是有用，其中一個藍圖可能指向
/v1/<routes>，另外一個指向/v2/<routes>。服務器當藍圖被初始化時，它可使用一個可選的
url_prefix參數，這個參數將被添加到藍圖上定義的全部路由上。此功能可用於實現API版本控制。cookie
from sanic.response import text
from sanic import Blueprint
blueprint_v1 = Blueprint('v1', url_prefix='/v1')
blueprint_v2 = Blueprint('v2', url_prefix='/v2')
@blueprint_v1.route('/')
async def api_v1_root(request):
return text('版本一')
@blueprint_v2.route('/')
async def api_v2_root(request):
return text('版本二')
- 註冊
from sanic import Sanic from blueprints import blueprint_v1, blueprint_v2 app = Sanic(__name__) app.blueprint(blueprint_v1, url_prefix='/v1') app.blueprint(blueprint_v2, url_prefix='/v2') app.run(host='0.0.0.0', port=8000, debug=True)
4. 配置
# 第一種方式
app = Sanic('myapp')
app.config.DB_NAME = 'appdb'
app.config.DB_USER = 'appuser'
# 第二種方式
db_settings = {
'DB_HOST': 'localhost',
'DB_NAME': 'appdb',
'DB_USER': 'appuser'
}
app.config.update(db_settings)
配置加載方式
從環境變量加載
# 任何由SANIC_定義的變量都將應用於sanic配置。例如，設置SANIC_REQUEST_TIMEOUT自動加載應用程序。你可使用load_cars將布爾值傳遞給Sanic構造函數來進行覆蓋。 app = Sanic(load_vars=False)
從對象加載
# 若是有不少配置參數而且它們有合理的默認值，將它們放置於模塊是有幫助的。 import myapp.default_settings app = Sanic('myapp') app.config.from_object(myapp.default_settings)從文件加載
# 一般狀況下，你想要從文件中加載配置參數。你能夠從from_file(/path/to/config_file)來加載配置參數。然而，這須要程序知道配置文件的位置，因此你能夠在環境變量中指定配置文件的路徑，並讓Sanic尋找配置文件並使用配置文件。 app = Sanic('myapp') app.config.from_envvar('MYAPP_SETTINGS') # 而後你能夠在MYAPP_SETTINGS環境設置下運行你的應用程序： $ MYAPP_SETTINGS=/path/to/config_file python3 myapp.py INFO: Goin' Fast @ http://0.0.0.0:8000 # 配置文件是常規的Python文件，運行它們只是爲了加載配置。這容許你使用任何正確的邏輯進行正確的配置。只要uppercase變量被添加到配置中，最多見的配置包括簡單的鍵值對： # config_file DB_HOST = 'localhost' DB_NAME = 'appdb' DB_USER = 'appuser'
5. COOKIES
響應的cookies能夠設置爲字典值，同時也有如下參數可用：app
expires(時間): cookie最後在客戶端瀏覽器上存在時間。path(字符串): Cookie的URL子集。默認爲/。comment(字符串): 註釋（元數據）。domain(字符串): 指定cookie有效的域。顯式指定的域必須始終以點開頭。max-age(數字): cookie應該存在的秒數。secure(布爾值): 指定cookie是否只能經過HTTPS發送。httponly(布爾值): 指定cookie是否能被Javascript讀取。
from sanic import Sanic
from sanic.response import text
app = Sanic()
@app.route("/get_cookie")
async def test(request):
test_cookie = request.cookies.get('test')
return text("Test cookie set to: {}".format(test_cookie))
@app.route("/set_cookie")
async def test(request):
response = text("There's a cookie up in this response")
response.cookies['id'] = 'It worked!'
# response.cookies['test']['domain'] = '.gotta-go-fast.com'
# response.cookies['test']['httponly'] = True
return response
@app.route("/del_cookie")
async def test(request):
response = text("刪除cookies")
# 此cookie將被設置爲0秒後過時
del response.cookies['kill_me']
# 這個cookie將在5秒內自動銷燬
response.cookies['short_life'] = 'Glad to be here'
response.cookies['short_life']['max-age'] = 5
del response.cookies['favorite_color']
# This cookie will remain unchanged
response.cookies['favorite_color'] = 'blue'
response.cookies['favorite_color'] = 'pink'
del response.cookies['favorite_color']
return response
if __name__ == '__main__':
app.run()
6.裝飾器添加
from functools import wraps
from sanic import Sanic
from sanic.response import json
app = Sanic()
def auth():
def decorator(f):
@wraps(f)
async def decorated_function(request, *args, **kwargs):
# 判斷是否經過 認證
is_authorized = True
if is_authorized:
# 經過認證
response = await f(request, *args, **kwargs)
return response
else:
# 未經過認證
return json({'status': 'not_authorized'}, 403)
return decorated_function
return decorator
@app.route("/")
@auth()
async def test(request):
return json({"status": 'authorized'})
if __name__ == '__main__':
app.run()
7. CBV 試圖寫法
from functools import wraps
from sanic.response import text, json, redirect
from sanic.views import HTTPMethodView, CompositionView
from sanic import Sanic
app = Sanic(__name__)
def auth():
def decorator(f):
@wraps(f)
async def decorated_function(request, *args, **kwargs):
# 判斷是否經過 認證
is_authorized = True
print("認證裝飾器")
if is_authorized:
# 經過認證
response = await f(request, *args, **kwargs)
return response
else:
# 未經過認證
return json({'status': 'not_authorized'}, 403)
return decorated_function
return decorator
# @app.route('/')
async def index(request):
url = app.url_for('Users', id=10) # 反向解析
print(url) # /user/10
return redirect(url) # 302
# 使用組成視圖 處理錯誤請求
view = CompositionView()
view.add(['GET'], index)
view.add(['POST', 'PUT'], lambda request: text('我沒有 post / put 方法'))
app.add_route(view, '/')
class Users(HTTPMethodView):
decorators = [auth(), ] # 加裝飾器
# 能夠不使用異步
async def get(self, request, id):
print(id)
return text('{}---OK'.format(id))
async def post(self, request,id):
return text("{}".format(request.form))
...
app.add_route(Users.as_view(), '/user/<id>')
if __name__ == '__main__':
app.run()
8.自定義協議
from sanic import Sanic
from sanic.response import text
from sanic.server import HttpProtocol
app = Sanic()
# 自定義 http 協議
class CustomHttpProtocol(HttpProtocol):
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
def write_response(self, response):
if isinstance(response, str):
response = text(response)
self.transport.write(
response.output(self.request.version)
)
self.transport.close()
@app.route('/')
async def string(request):
return 'string'
@app.route('/1')
async def response(request):
return text('response')
app.run(protocol=CustomHttpProtocol) # 指定協議
9.SSL 證書
import ssl
ctx = ssl.create_default_context(purpose=ssl.Purpose.CLIENT_AUTH)
ctx.load_cert_chain("/path/to/cert", keyfile="/path/to/keyfile")
app.run(host="0.0.0.0", port=8443, ssl=ctx)
# 你還能夠將證書和密鑰的位置作爲自檢進行傳遞。
ssl = {'cert': "/path/to/cert", 'key': "/path/to/keyfile"}
app.run(host="0.0.0.0", port=5000, ssl=ssl)
10.命令行啓動
- 須要指定 server.py 文件爲啓動文件
python -m sanic server.app --host=0.0.0.0 --port=80 --workers=4
相關文章
- 1. Sanic 基礎應用 -1
- 2. 《雲計算》Shell的基礎應用（2）
- 3. Android 應用程序基礎知識（2）
- 4. 2、HTML基礎設計與應用
- 5. Linux應用編程基礎--（5）多進程編程基礎2
- 6. 基於Sanic的微服務基礎架構
- 7. 基於sanic的微服務基礎架構
- 8. sanic-jwt 的使用
- 9. 2、基礎使用
- 10. Sanic教程
- 更多相關文章...
- • R 基礎運算 - R 語言教程
- • R 語言基礎 - R 語言教程
- • TiDB 在摩拜單車在線數據業務的應用和實踐
- • Kotlin學習（二）基本類型
相關標籤/搜索
每日一句
-
每一个你不满意的现在，都有一个你没有努力的曾经。
歡迎關注本站公眾號,獲取更多信息
相關文章
- 1. Sanic 基礎應用 -1
- 2. 《雲計算》Shell的基礎應用（2）
- 3. Android 應用程序基礎知識（2）
- 4. 2、HTML基礎設計與應用
- 5. Linux應用編程基礎--（5）多進程編程基礎2
- 6. 基於Sanic的微服務基礎架構
- 7. 基於sanic的微服務基礎架構
- 8. sanic-jwt 的使用
- 9. 2、基礎使用
- 10. Sanic教程