falcon
4.0.2
Falcon เป็นเฟรมเวิร์ก ASGI/WSGI แบบเรียบง่ายสำหรับการสร้าง REST API และไมโครเซอร์วิสที่มีความสำคัญต่อภารกิจ โดยมุ่งเน้นไปที่ความน่าเชื่อถือ ความถูกต้อง และประสิทธิภาพในระดับต่างๆ
เมื่อพูดถึงการสร้าง HTTP API เฟรมเวิร์กอื่นๆ จะถ่วงน้ำหนักคุณด้วยการพึ่งพามากมายและนามธรรมที่ไม่จำเป็น Falcon ก้าวเข้าสู่การไล่ล่าด้วยการออกแบบที่สะอาดตาซึ่งรวมเอาสไตล์สถาปัตยกรรม HTTP และ REST
แอป Falcon ทำงานร่วมกับเซิร์ฟเวอร์ WSGI หรือ ASGI ใดก็ได้ และทำงานเหมือนแชมป์ภายใต้ CPython 3.8+ และ PyPy 3.8+
“ฟอลคอนแข็งแกร่งมากและรวดเร็ว”
“เราใช้ Falcon มาแทนที่ [เฟรมเวิร์กอื่น] และเราชอบประสิทธิภาพ (เร็วกว่าสามเท่า) และขนาดฐานโค้ด (ง่ายกว่าครึ่งหนึ่งของโค้ด [ดั้งเดิม] ของเราอย่างง่ายดาย)”
"ฉันชอบ #falconframework! สะอาดตาและเรียบง่ายสุดๆ ในที่สุดฉันก็มีความเร็วและความยืดหยุ่นที่ต้องการแล้ว!"
"จนถึงตอนนี้ Falcon ดูดีมาก ฉันแฮ็กการทดสอบด่วนสำหรับเซิร์ฟเวอร์เล็กๆ ของฉันและเร็วขึ้น ~40% โดยใช้เวลาทำงานเพียง 20 นาที"
"ในที่สุดฉันก็รู้สึกเหมือนว่าฉันกำลังพูดถึง HTTP โดยไม่มีอะไรอยู่ตรงกลาง Falcon ดูเหมือนเป็นคำขอของแบ็กเอนด์"
"ซอร์สโค้ดของ Falcon ดีมาก ฉันเกือบจะชอบมันมากกว่าเอกสารประกอบ โดยทั่วไปแล้วมันไม่ผิดเลย"
"มีเฟรมเวิร์กอื่นใดอีกบ้างที่รวมการสนับสนุนสำหรับ 786 TRY IT NOW"
ฟอลคอนพยายามทำให้น้อยที่สุดเท่าที่จะเป็นไปได้ในขณะที่ยังคงมีประสิทธิภาพสูงอยู่
asyncio ดั้งเดิมFalcon ช่วยคุณสร้างแอปที่ยอดเยี่ยมหรือไม่? แสดงการสนับสนุนของคุณวันนี้ด้วยการบริจาคครั้งเดียวหรือเป็นผู้อุปถัมภ์ ผู้สนับสนุนจะได้รับอุปกรณ์เจ๋งๆ โอกาสในการโปรโมตแบรนด์ของตนไปยังนักพัฒนา Python และจัดลำดับความสำคัญของการสนับสนุน
ขอบคุณ!
ในที่สุดความสมบูรณ์แบบก็มาถึงไม่ใช่เมื่อไม่มีอะไรจะเพิ่มเติมอีกต่อไป แต่เมื่อไม่มีอะไรที่จะเอาออกไปอีกต่อไป
- อองตวน เดอ แซงเตกซูเปรี
เราออกแบบ Falcon เพื่อรองรับความต้องการที่ต้องการของไมโครเซอร์วิสขนาดใหญ่และแบ็กเอนด์แอปที่ตอบสนอง Falcon เสริมเฟรมเวิร์กเว็บ Python ทั่วไปโดยมอบประสิทธิภาพแบบ Bare Metal ความน่าเชื่อถือ และความยืดหยุ่นในทุกที่ที่คุณต้องการ
เชื่อถือได้. เราใช้ความพยายามอย่างเต็มที่เพื่อหลีกเลี่ยงการแนะนำการเปลี่ยนแปลงที่เสียหาย และเมื่อเราดำเนินการแล้ว การเปลี่ยนแปลงดังกล่าวจะได้รับการบันทึกไว้อย่างครบถ้วนและนำมาใช้เท่านั้น (ในจิตวิญญาณของ SemVer) โดยมีการเพิ่มเวอร์ชันหลักเท่านั้น รหัสได้รับการทดสอบอย่างเข้มงวดด้วยอินพุตจำนวนมาก และเราต้องการความครอบคลุม 100% ตลอดเวลา Falcon ไม่มีการพึ่งพานอกไลบรารีมาตรฐาน ซึ่งช่วยลดพื้นที่การโจมตีของแอปของคุณ ในขณะเดียวกันก็หลีกเลี่ยงข้อบกพร่องทางสกรรมกริยาและทำลายการเปลี่ยนแปลง
แก้ไขข้อบกพร่องได้ ฟอลคอนหลบเลี่ยงเวทมนตร์ เป็นเรื่องง่ายที่จะบอกได้ว่าอินพุตใดนำไปสู่เอาต์พุตใด ข้อยกเว้นที่ไม่สามารถจัดการได้จะไม่ถูกห่อหุ้มหรือปิดบังไว้ พฤติกรรมที่น่าประหลาดใจที่อาจเกิดขึ้น เช่น การแยกวิเคราะห์เนื้อหาคำขออัตโนมัติ ได้รับการบันทึกไว้อย่างดีและปิดใช้งานตามค่าเริ่มต้น สุดท้ายนี้ เมื่อพูดถึงกรอบงาน เราจะดูแลให้เส้นทางลอจิกเรียบง่ายและเข้าใจได้ ทั้งหมดนี้ทำให้ง่ายต่อการให้เหตุผลเกี่ยวกับโค้ดและแก้ไขข้อบกพร่องของ Edge Case ในการปรับใช้ขนาดใหญ่
เร็ว. ฮาร์ดแวร์เดียวกัน คำขอเพิ่มเติม Falcon ตอบกลับคำขอได้เร็วกว่าเฟรมเวิร์ก Python ยอดนิยมอื่นๆ เช่น Django และ Flask อย่างเห็นได้ชัด เพื่อเพิ่มความเร็วเป็นพิเศษ Falcon จะคอมไพล์ตัวเองด้วย Cython เมื่อพร้อมใช้งาน และยังทำงานได้ดีกับ PyPy กำลังพิจารณาที่จะย้ายไปยังภาษาการเขียนโปรแกรมอื่นหรือไม่? เกณฑ์มาตรฐานด้วย Falcon+PyPy ก่อน!
ยืดหยุ่นได้. Falcon ฝากการตัดสินใจและรายละเอียดการใช้งานไว้มากมายให้กับคุณซึ่งเป็นผู้พัฒนา API สิ่งนี้ให้อิสระแก่คุณในการปรับแต่งและปรับแต่งการใช้งานของคุณ นอกจากนี้ยังช่วยให้คุณเข้าใจแอปของคุณในระดับที่ลึกยิ่งขึ้น ทำให้ง่ายต่อการปรับแต่ง ดีบัก และปรับโครงสร้างใหม่ในระยะยาว การออกแบบที่เรียบง่ายของ Falcon มอบพื้นที่สำหรับสมาชิกชุมชน Python เพื่อสร้างสรรค์สิ่งใหม่ๆ บนโปรแกรมเสริม Falcon และแพ็คเกจเสริมอย่างอิสระ
Falcon ถูกใช้ทั่วโลกโดยองค์กรจำนวนมากขึ้นเรื่อยๆ รวมถึง:
หากคุณกำลังใช้เฟรมเวิร์ก Falcon สำหรับชุมชนหรือโครงการเชิงพาณิชย์ โปรดพิจารณาเพิ่มข้อมูลของคุณลงในวิกิของเราภายใต้ ใครกำลังใช้ Falcon?
มีโปรแกรมเสริม เทมเพลต และแพ็คเกจเสริมของ Falcon จำนวนมากให้เลือกใช้ในโปรเจ็กต์ของคุณ เราได้แสดงรายการสิ่งเหล่านี้ไว้หลายรายการในวิกิ Falcon เป็นจุดเริ่มต้น แต่คุณอาจต้องการค้นหา PyPI เพื่อหาแหล่งข้อมูลเพิ่มเติม
ชุมชน Falconry บน Gitter เป็นสถานที่ที่ดีเยี่ยมในการถามคำถามและแบ่งปันความคิดของคุณ คุณจะพบเราได้ในเหยี่ยว/ผู้ใช้ นอกจากนี้เรายังมีห้องเหยี่ยว/นักพัฒนาสำหรับหารือเกี่ยวกับการออกแบบและพัฒนาเฟรมเวิร์กด้วย
ตามหลักจรรยาบรรณของเรา เราคาดหวังให้ทุกคนที่มีส่วนร่วมในการอภิปรายในชุมชนกระทำการอย่างมืออาชีพ และเป็นผู้นำเป็นตัวอย่างในการสนับสนุนการสนทนาที่สร้างสรรค์ แต่ละคนในชุมชนมีหน้าที่รับผิดชอบในการสร้างวัฒนธรรมเชิงบวก สร้างสรรค์ และมีประสิทธิผล
PyPy เป็นวิธีที่เร็วที่สุดในการรันแอป Falcon ของคุณ รองรับ PyPy3.8+ ตั้งแต่ PyPy v7.3.7+
$ pip install falconหรือหากต้องการติดตั้งเบต้าหรือรีลีสล่าสุด ถ้ามี:
$ pip install --pre falconFalcon ยังรองรับ CPython 3.8+ อย่างสมบูรณ์
Falcon เวอร์ชันเสถียรล่าสุดสามารถติดตั้งได้โดยตรงจาก PyPI:
$ pip install falconหรือหากต้องการติดตั้งเบต้าหรือรีลีสล่าสุด ถ้ามี:
$ pip install --pre falconเพื่อเพิ่มความเร็วเป็นพิเศษ Falcon จะคอมไพล์ตัวเองด้วย Cython โดยอัตโนมัติภายใต้โปรแกรมติดตั้งที่รองรับ PEP 517
เพื่อความสะดวกของคุณ ล้อที่มีไบนารีที่คอมไพล์ไว้ล่วงหน้าจะพร้อมใช้งานจาก PyPI สำหรับแพลตฟอร์มทั่วไปส่วนใหญ่ แม้ว่าจะไม่มีการสร้างไบนารีสำหรับแพลตฟอร์มที่คุณเลือก แต่ pip ก็จะเลือกวงล้อ Python ที่แท้จริง คุณยังสามารถไซโทไนซ์ Falcon สำหรับสภาพแวดล้อมของคุณได้ ดูเอกสารการติดตั้งของเราสำหรับข้อมูลเพิ่มเติมเกี่ยวกับตัวเลือกขั้นสูงนี้และตัวเลือกขั้นสูงอื่น ๆ
Falcon ไม่จำเป็นต้องติดตั้งแพ็คเกจอื่นใด
Falcon พูด WSGI (หรือ ASGI ดูด้านล่าง) เพื่อให้บริการแอป Falcon คุณจะต้องมีเซิร์ฟเวอร์ WSGI Gunicorn และ uWSGI เป็นกลุ่มที่ได้รับความนิยมมากกว่า แต่สิ่งที่สามารถโหลดแอป WSGI ได้ก็สามารถทำได้
$ pip install [gunicorn | uwsgi]เพื่อให้บริการแอป Falcon ASGI คุณจะต้องมีเซิร์ฟเวอร์ ASGI Uvicorn เป็นตัวเลือกยอดนิยม:
$ pip install uvicornFalcon ใช้งานบน GitHub ทำให้ง่ายต่อการเรียกดู ดาวน์โหลด แยก ฯลฯ ยินดีรับคำขอดึงเสมอ! นอกจากนี้ โปรดอย่าลืมติดดาวโปรเจ็กต์นี้ด้วยถ้ามันทำให้คุณมีความสุข -
เมื่อคุณโคลน repo หรือดาวน์โหลด tarball จาก GitHub แล้ว คุณสามารถติดตั้ง Falcon ได้ดังนี้:
$ cd falcon
$ pip install .หรือ หากคุณต้องการแก้ไขโค้ด ขั้นแรกให้แยก repo หลัก โคลน fork ไปที่เดสก์ท็อปของคุณ จากนั้นเรียกใช้สิ่งต่อไปนี้เพื่อติดตั้งโดยใช้การลิงก์สัญลักษณ์ เพื่อที่เมื่อคุณเปลี่ยนโค้ด การเปลี่ยนแปลงจะพร้อมใช้งานโดยอัตโนมัติ แอปของคุณโดยไม่ต้องติดตั้งแพ็คเกจใหม่:
$ cd falcon
$ FALCON_DISABLE_CYTHON=Y pip install -e .คุณสามารถทดสอบการเปลี่ยนแปลงในเฟรมเวิร์ก Falcon ได้ด้วยตนเองโดยสลับไปยังไดเร็กทอรีของ repo ที่ลอกแบบมา จากนั้นรัน pytest:
$ cd falcon
$ pip install -r requirements/tests
$ pytest testsหรือหากต้องการรันชุดการทดสอบเริ่มต้น:
$ pip install tox && toxโปรดดูไฟล์ tox.ini สำหรับรายการสภาพแวดล้อมทั้งหมดที่มีอยู่
เอกสารในฐานโค้ด Falcon นั้นค่อนข้างกว้างขวาง และเราขอแนะนำให้ใช้งาน REPL ต่อไปในขณะที่เรียนรู้เฟรมเวิร์ก เพื่อให้คุณสามารถสืบค้นโมดูลและคลาสต่างๆ เมื่อคุณมีคำถาม
เอกสารออนไลน์มีอยู่ที่: https://falcon.readthedocs.io
คุณสามารถสร้างเอกสารเดียวกันภายในเครื่องได้ดังนี้:
$ pip install tox && tox -e docsเมื่อสร้างเอกสารแล้ว คุณสามารถดูได้โดยเปิดหน้าดัชนีต่อไปนี้ในเบราว์เซอร์ของคุณ บน OS X ทำได้ง่ายเพียง:
$ เปิดเอกสาร/_build/html/index.html
หรือบนลินุกซ์:
$ xdg-open docs/_build/html/index.htmlนี่คือตัวอย่างง่ายๆ ที่สร้างสรรค์ซึ่งแสดงวิธีสร้างแอป WSGI ที่ใช้ Falcon (เวอร์ชัน ASGI จะรวมอยู่ด้านล่าง):
# examples/things.py
# Let's get this party started!
from wsgiref . simple_server import make_server
import falcon
# Falcon follows the REST architectural style, meaning (among
# other things) that you think in terms of resources and state
# transitions, which map to HTTP verbs.
class ThingsResource :
def on_get ( self , req , resp ):
"""Handles GET requests"""
resp . status = falcon . HTTP_200 # This is the default status
resp . content_type = falcon . MEDIA_TEXT # Default is JSON, so override
resp . text = ( ' n Two things awe me most, the starry sky '
'above me and the moral law within me. n '
' n '
' ~ Immanuel Kant n n ' )
# falcon.App instances are callable WSGI apps...
# in larger applications the app is created in a separate file
app = falcon . App ()
# Resources are represented by long-lived class instances
things = ThingsResource ()
# things will handle all requests to the '/things' URL path
app . add_route ( '/things' , things )
if __name__ == '__main__' :
with make_server ( '' , 8000 , app ) as httpd :
print ( 'Serving on port 8000...' )
# Serve until process is killed
httpd . serve_forever ()คุณสามารถรันตัวอย่างข้างต้นได้โดยตรงโดยใช้เซิร์ฟเวอร์ wsgiref ที่รวมอยู่:
$ pip install falcon
$ python things.pyจากนั้นในเทอร์มินัลอื่น:
$ curl localhost:8000/thingsตัวอย่างเวอร์ชัน ASGI คล้ายกัน:
# examples/things_asgi.py
import falcon
import falcon . asgi
# Falcon follows the REST architectural style, meaning (among
# other things) that you think in terms of resources and state
# transitions, which map to HTTP verbs.
class ThingsResource :
async def on_get ( self , req , resp ):
"""Handles GET requests"""
resp . status = falcon . HTTP_200 # This is the default status
resp . content_type = falcon . MEDIA_TEXT # Default is JSON, so override
resp . text = ( ' n Two things awe me most, the starry sky '
'above me and the moral law within me. n '
' n '
' ~ Immanuel Kant n n ' )
# falcon.asgi.App instances are callable ASGI apps...
# in larger applications the app is created in a separate file
app = falcon . asgi . App ()
# Resources are represented by long-lived class instances
things = ThingsResource ()
# things will handle all requests to the '/things' URL path
app . add_route ( '/things' , things )คุณสามารถรันเวอร์ชัน ASGI ด้วย uvicorn หรือเซิร์ฟเวอร์ ASGI อื่นๆ ได้:
$ pip install falcon uvicorn
$ uvicorn things_asgi:appต่อไปนี้คือตัวอย่างที่เกี่ยวข้องเพิ่มเติมซึ่งสาธิตการอ่านส่วนหัวและพารามิเตอร์การสืบค้น การจัดการข้อผิดพลาด และการทำงานกับเนื้อหาคำขอและการตอบกลับ โปรดทราบว่าตัวอย่างนี้ถือว่ามีการติดตั้งแพ็คเกจคำขอแล้ว
(สำหรับแอป ASGI ที่เทียบเท่า โปรดดู: ตัวอย่างที่ซับซ้อนมากขึ้น (ASGI))
# examples/things_advanced.py
import json
import logging
import uuid
from wsgiref import simple_server
import falcon
import requests
class StorageEngine :
def get_things ( self , marker , limit ):
return [{ 'id' : str ( uuid . uuid4 ()), 'color' : 'green' }]
def add_thing ( self , thing ):
thing [ 'id' ] = str ( uuid . uuid4 ())
return thing
class StorageError ( Exception ):
@ staticmethod
def handle ( ex , req , resp , params ):
# TODO: Log the error, clean up, etc. before raising
raise falcon . HTTPInternalServerError ()
class SinkAdapter :
engines = {
'ddg' : 'https://duckduckgo.com' ,
'y' : 'https://search.yahoo.com/search' ,
}
def __call__ ( self , req , resp , engine ):
url = self . engines [ engine ]
params = { 'q' : req . get_param ( 'q' , True )}
result = requests . get ( url , params = params )
resp . status = str ( result . status_code ) + ' ' + result . reason
resp . content_type = result . headers [ 'content-type' ]
resp . text = result . text
class AuthMiddleware :
def process_request ( self , req , resp ):
token = req . get_header ( 'Authorization' )
account_id = req . get_header ( 'Account-ID' )
challenges = [ 'Token type="Fernet"' ]
if token is None :
description = ( 'Please provide an auth token '
'as part of the request.' )
raise falcon . HTTPUnauthorized ( title = 'Auth token required' ,
description = description ,
challenges = challenges ,
href = 'http://docs.example.com/auth' )
if not self . _token_is_valid ( token , account_id ):
description = ( 'The provided auth token is not valid. '
'Please request a new token and try again.' )
raise falcon . HTTPUnauthorized ( title = 'Authentication required' ,
description = description ,
challenges = challenges ,
href = 'http://docs.example.com/auth' )
def _token_is_valid ( self , token , account_id ):
return True # Suuuuuure it's valid...
class RequireJSON :
def process_request ( self , req , resp ):
if not req . client_accepts_json :
raise falcon . HTTPNotAcceptable (
description = 'This API only supports responses encoded as JSON.' ,
href = 'http://docs.examples.com/api/json' )
if req . method in ( 'POST' , 'PUT' ):
if 'application/json' not in req . content_type :
raise falcon . HTTPUnsupportedMediaType (
title = 'This API only supports requests encoded as JSON.' ,
href = 'http://docs.examples.com/api/json' )
class JSONTranslator :
# NOTE: Normally you would simply use req.media and resp.media for
# this particular use case; this example serves only to illustrate
# what is possible.
def process_request ( self , req , resp ):
# req.stream corresponds to the WSGI wsgi.input environ variable,
# and allows you to read bytes from the request body.
#
# See also: PEP 3333
if req . content_length in ( None , 0 ):
# Nothing to do
return
body = req . stream . read ()
if not body :
raise falcon . HTTPBadRequest ( title = 'Empty request body' ,
description = 'A valid JSON document is required.' )
try :
req . context . doc = json . loads ( body . decode ( 'utf-8' ))
except ( ValueError , UnicodeDecodeError ):
description = ( 'Could not decode the request body. The '
'JSON was incorrect or not encoded as '
'UTF-8.' )
raise falcon . HTTPBadRequest ( title = 'Malformed JSON' ,
description = description )
def process_response ( self , req , resp , resource , req_succeeded ):
if not hasattr ( resp . context , 'result' ):
return
resp . text = json . dumps ( resp . context . result )
def max_body ( limit ):
def hook ( req , resp , resource , params ):
length = req . content_length
if length is not None and length > limit :
msg = ( 'The size of the request is too large. The body must not '
'exceed ' + str ( limit ) + ' bytes in length.' )
raise falcon . HTTPContentTooLarge (
title = 'Request body is too large' , description = msg )
return hook
class ThingsResource :
def __init__ ( self , db ):
self . db = db
self . logger = logging . getLogger ( 'thingsapp.' + __name__ )
def on_get ( self , req , resp , user_id ):
marker = req . get_param ( 'marker' ) or ''
limit = req . get_param_as_int ( 'limit' ) or 50
try :
result = self . db . get_things ( marker , limit )
except Exception as ex :
self . logger . error ( ex )
description = ( 'Aliens have attacked our base! We will '
'be back as soon as we fight them off. '
'We appreciate your patience.' )
raise falcon . HTTPServiceUnavailable (
title = 'Service Outage' ,
description = description ,
retry_after = 30 )
# NOTE: Normally you would use resp.media for this sort of thing;
# this example serves only to demonstrate how the context can be
# used to pass arbitrary values between middleware components,
# hooks, and resources.
resp . context . result = result
resp . set_header ( 'Powered-By' , 'Falcon' )
resp . status = falcon . HTTP_200
@ falcon . before ( max_body ( 64 * 1024 ))
def on_post ( self , req , resp , user_id ):
try :
doc = req . context . doc
except AttributeError :
raise falcon . HTTPBadRequest (
title = 'Missing thing' ,
description = 'A thing must be submitted in the request body.' )
proper_thing = self . db . add_thing ( doc )
resp . status = falcon . HTTP_201
resp . location = '/%s/things/%s' % ( user_id , proper_thing [ 'id' ])
# Configure your WSGI server to load "things.app" (app is a WSGI callable)
app = falcon . App ( middleware = [
AuthMiddleware (),
RequireJSON (),
JSONTranslator (),
])
db = StorageEngine ()
things = ThingsResource ( db )
app . add_route ( '/{user_id}/things' , things )
# If a responder ever raises an instance of StorageError, pass control to
# the given handler.
app . add_error_handler ( StorageError , StorageError . handle )
# Proxy some things to another service; this example shows how you might
# send parts of an API off to a legacy system that hasn't been upgraded
# yet, or perhaps is a single cluster that all data centers have to share.
sink = SinkAdapter ()
app . add_sink ( sink , r'/search/(?P<engine>ddg|y)Z' )
# Useful for debugging problems in your API; works with pdb.set_trace(). You
# can also use Gunicorn to host your app. Gunicorn can be configured to
# auto-restart workers when it detects a code change, and it also works
# with pdb.
if __name__ == '__main__' :
httpd = simple_server . make_server ( '127.0.0.1' , 8000 , app )
httpd . serve_forever ()โค้ดนี้ใช้ wsgiref อีกครั้ง แต่คุณสามารถเรียกใช้ตัวอย่างข้างต้นได้โดยใช้เซิร์ฟเวอร์ WSGI ใดก็ได้ เช่น uWSGI หรือ Gunicorn ตัวอย่างเช่น:
$ pip install requests gunicorn
$ gunicorn things:appบน Windows คุณสามารถเรียกใช้ Gunicorn และ uWSGI ผ่าน WSL หรือคุณอาจลองใช้ Waitress:
$ pip install requests waitress
$ waitress-serve --port=8000 things:appหากต้องการทดสอบตัวอย่างนี้ ให้เปิดเทอร์มินัลอื่นแล้วรัน:
$ http localhost:8000/1/things authorization:custom-token คุณยังสามารถดูการกำหนดค่าแอปพลิเคชันจาก CLI ผ่านสคริปต์ falcon-inspect-app ที่มาพร้อมกับเฟรมเวิร์ก:
falcon-inspect-app things_advanced:appนี่คือแอปเวอร์ชัน ASGI จากด้านบน โปรดทราบว่าจะใช้แพ็คเกจ httpx แทนคำขอ
# examples/things_advanced_asgi.py
import json
import logging
import uuid
import falcon
import falcon . asgi
import httpx
class StorageEngine :
async def get_things ( self , marker , limit ):
return [{ 'id' : str ( uuid . uuid4 ()), 'color' : 'green' }]
async def add_thing ( self , thing ):
thing [ 'id' ] = str ( uuid . uuid4 ())
return thing
class StorageError ( Exception ):
@ staticmethod
async def handle ( ex , req , resp , params ):
# TODO: Log the error, clean up, etc. before raising
raise falcon . HTTPInternalServerError ()
class SinkAdapter :
engines = {
'ddg' : 'https://duckduckgo.com' ,
'y' : 'https://search.yahoo.com/search' ,
}
async def __call__ ( self , req , resp , engine ):
url = self . engines [ engine ]
params = { 'q' : req . get_param ( 'q' , True )}
async with httpx . AsyncClient () as client :
result = await client . get ( url , params = params )
resp . status = result . status_code
resp . content_type = result . headers [ 'content-type' ]
resp . text = result . text
class AuthMiddleware :
async def process_request ( self , req , resp ):
token = req . get_header ( 'Authorization' )
account_id = req . get_header ( 'Account-ID' )
challenges = [ 'Token type="Fernet"' ]
if token is None :
description = ( 'Please provide an auth token '
'as part of the request.' )
raise falcon . HTTPUnauthorized ( title = 'Auth token required' ,
description = description ,
challenges = challenges ,
href = 'http://docs.example.com/auth' )
if not self . _token_is_valid ( token , account_id ):
description = ( 'The provided auth token is not valid. '
'Please request a new token and try again.' )
raise falcon . HTTPUnauthorized ( title = 'Authentication required' ,
description = description ,
challenges = challenges ,
href = 'http://docs.example.com/auth' )
def _token_is_valid ( self , token , account_id ):
return True # Suuuuuure it's valid...
class RequireJSON :
async def process_request ( self , req , resp ):
if not req . client_accepts_json :
raise falcon . HTTPNotAcceptable (
description = 'This API only supports responses encoded as JSON.' ,
href = 'http://docs.examples.com/api/json' )
if req . method in ( 'POST' , 'PUT' ):
if 'application/json' not in req . content_type :
raise falcon . HTTPUnsupportedMediaType (
description = 'This API only supports requests encoded as JSON.' ,
href = 'http://docs.examples.com/api/json' )
class JSONTranslator :
# NOTE: Normally you would simply use req.get_media() and resp.media for
# this particular use case; this example serves only to illustrate
# what is possible.
async def process_request ( self , req , resp ):
# NOTE: Test explicitly for 0, since this property could be None in
# the case that the Content-Length header is missing (in which case we
# can't know if there is a body without actually attempting to read
# it from the request stream.)
if req . content_length == 0 :
# Nothing to do
return
body = await req . stream . read ()
if not body :
raise falcon . HTTPBadRequest ( title = 'Empty request body' ,
description = 'A valid JSON document is required.' )
try :
req . context . doc = json . loads ( body . decode ( 'utf-8' ))
except ( ValueError , UnicodeDecodeError ):
description = ( 'Could not decode the request body. The '
'JSON was incorrect or not encoded as '
'UTF-8.' )
raise falcon . HTTPBadRequest ( title = 'Malformed JSON' ,
description = description )
async def process_response ( self , req , resp , resource , req_succeeded ):
if not hasattr ( resp . context , 'result' ):
return
resp . text = json . dumps ( resp . context . result )
def max_body ( limit ):
async def hook ( req , resp , resource , params ):
length = req . content_length
if length is not None and length > limit :
msg = ( 'The size of the request is too large. The body must not '
'exceed ' + str ( limit ) + ' bytes in length.' )
raise falcon . HTTPContentTooLarge (
title = 'Request body is too large' , description = msg )
return hook
class ThingsResource :
def __init__ ( self , db ):
self . db = db
self . logger = logging . getLogger ( 'thingsapp.' + __name__ )
async def on_get ( self , req , resp , user_id ):
marker = req . get_param ( 'marker' ) or ''
limit = req . get_param_as_int ( 'limit' ) or 50
try :
result = await self . db . get_things ( marker , limit )
except Exception as ex :
self . logger . error ( ex )
description = ( 'Aliens have attacked our base! We will '
'be back as soon as we fight them off. '
'We appreciate your patience.' )
raise falcon . HTTPServiceUnavailable (
title = 'Service Outage' ,
description = description ,
retry_after = 30 )
# NOTE: Normally you would use resp.media for this sort of thing;
# this example serves only to demonstrate how the context can be
# used to pass arbitrary values between middleware components,
# hooks, and resources.
resp . context . result = result
resp . set_header ( 'Powered-By' , 'Falcon' )
resp . status = falcon . HTTP_200
@ falcon . before ( max_body ( 64 * 1024 ))
async def on_post ( self , req , resp , user_id ):
try :
doc = req . context . doc
except AttributeError :
raise falcon . HTTPBadRequest (
title = 'Missing thing' ,
description = 'A thing must be submitted in the request body.' )
proper_thing = await self . db . add_thing ( doc )
resp . status = falcon . HTTP_201
resp . location = '/%s/things/%s' % ( user_id , proper_thing [ 'id' ])
# The app instance is an ASGI callable
app = falcon . asgi . App ( middleware = [
# AuthMiddleware(),
RequireJSON (),
JSONTranslator (),
])
db = StorageEngine ()
things = ThingsResource ( db )
app . add_route ( '/{user_id}/things' , things )
# If a responder ever raises an instance of StorageError, pass control to
# the given handler.
app . add_error_handler ( StorageError , StorageError . handle )
# Proxy some things to another service; this example shows how you might
# send parts of an API off to a legacy system that hasn't been upgraded
# yet, or perhaps is a single cluster that all data centers have to share.
sink = SinkAdapter ()
app . add_sink ( sink , r'/search/(?P<engine>ddg|y)Z' )คุณสามารถรันเวอร์ชัน ASGI กับเซิร์ฟเวอร์ ASGI ใดก็ได้ เช่น uvicorn:
$ pip install falcon httpx uvicorn
$ uvicorn things_advanced_asgi:appขอขอบคุณที่ให้ความสนใจในโครงการนี้! เรายินดีรับคำขอดึงจากนักพัฒนาทุกระดับทักษะ ในการเริ่มต้น เพียงแยกสาขาหลักบน GitHub ไปยังบัญชีส่วนตัวของคุณ จากนั้นจึงโคลนทางแยกไปยังสภาพแวดล้อมการพัฒนาของคุณ
หากคุณต้องการมีส่วนร่วมแต่ยังไม่มีสิ่งใดอยู่ในใจ เราขอเชิญชวนให้คุณดูปัญหาต่างๆ ที่ระบุไว้ในเหตุการณ์สำคัญครั้งต่อไปของเรา หากคุณเห็นสิ่งที่คุณต้องการดำเนินการ โปรดแสดงความคิดเห็นสั้นๆ เพื่อที่เราจะได้ไม่จบลงด้วยความพยายามซ้ำซ้อน ขอบคุณล่วงหน้า!
โปรดทราบว่าผู้ร่วมให้ข้อมูลและผู้ดูแลโครงการนี้ทั้งหมดอยู่ภายใต้หลักจรรยาบรรณของเรา
ก่อนที่จะส่งคำขอดึง โปรดตรวจสอบให้แน่ใจว่าคุณได้เพิ่ม/อัปเดตการทดสอบที่เหมาะสม (และการทดสอบที่มีอยู่ทั้งหมดยังคงผ่านการทดสอบตามการเปลี่ยนแปลงของคุณ) และสไตล์การเขียนโค้ดของคุณเป็นไปตาม PEP 8 และไม่ทำให้ไพเฟลกบ่น
ข้อความที่คอมมิตควรได้รับการจัดรูปแบบโดยใช้แบบแผน AngularJS
ความคิดเห็นเป็นไปตามคำแนะนำสไตล์ของ Google โดยมีข้อกำหนดเพิ่มเติมในการเติมความคิดเห็นแบบอินไลน์โดยใช้ชื่อเล่น GitHub ของคุณและคำนำหน้าที่เหมาะสม:
ผู้ดูแลโครงการ Falcon หลัก ได้แก่:
โปรดอย่าลังเลที่จะติดต่อเราหากคุณมีคำถามใดๆ หรือต้องการความช่วยเหลือเล็กน้อยในการเริ่มต้น คุณจะพบเราได้ใน Falconry/dev บน Gitter
ดูเพิ่มเติมที่: CONTRIBUTING.md
ลิขสิทธิ์ 2013-2024 โดยบุคคลและองค์กรตามที่ระบุไว้ในไฟล์ต้นฉบับแต่ละไฟล์
ได้รับอนุญาตภายใต้ Apache License เวอร์ชัน 2.0 ("ใบอนุญาต"); คุณไม่สามารถใช้ส่วนใด ๆ ของเฟรมเวิร์ก Falcon ยกเว้นการปฏิบัติตามใบอนุญาต ผู้ร่วมให้ข้อมูลตกลงที่จะอนุญาตการทำงานของตนภายใต้ใบอนุญาตเดียวกัน คุณสามารถขอรับสำเนาใบอนุญาตได้ที่ http://www.apache.org/licenses/LICENSE-2.0
เว้นแต่กฎหมายที่ใช้บังคับกำหนดหรือตกลงเป็นลายลักษณ์อักษร ซอฟต์แวร์ที่เผยแพร่ภายใต้ใบอนุญาตนี้จะถูกแจกจ่าย "ตามที่เป็น" โดยไม่มีการรับประกันหรือเงื่อนไขใดๆ ทั้งโดยชัดแจ้งหรือโดยนัย ดูใบอนุญาตสำหรับภาษาเฉพาะที่ควบคุมการอนุญาตและข้อจำกัดภายใต้ใบอนุญาต
