FastAPI по суті є обгорткою для Starlette. Щоб повністю зрозуміти FastAPI, спершу потрібно зрозуміти Starlette.
1. Протокол ASGI
Uvicorn взаємодіє з ASGI-додатками через спільний інтерфейс. Додаток може надсилати та отримувати інформацію через Uvicorn, реалізуючи наступний код:
async def app(scope, receive, send):
# Найпростіший ASGI-додаток
assert scope['type'] == 'http'
await send({
'type': 'http.response.start',
'status': 200,
'headers': [
[b'content-type', b'text/plain'],
]
})
await send({
'type': 'http.response.body',
'body': b'Hello, world!',
})
if __name__ == "__main__":
# Сервіс Uvicorn
import uvicorn
uvicorn.run(app, host="127.0.0.1", port=5000, log_level="info")
2. Starlette
Щоб запустити Starlette з Uvicorn, використовуйте наступний код:
from starlette.applications import Starlette
from starlette.middleware.gzip import GZipMiddleware
app: Starlette = Starlette()
@app.route("/")
def demo_route() -> None: pass
@app.websocket_route("/")
def demo_websocket_route() -> None: pass
@app.add_exception_handlers(404)
def not_found_route() -> None: pass
@app.on_event("startup")
def startup_event_demo() -> None: pass
@app.on_event("shutdown")
def shutdown_event_demo() -> None: pass
app.add_middleware(GZipMiddleware)
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="127.0.0.1", port=5000)
Цей код ініціалізує Starlette, реєструє маршрути, обробники винятків, події та middleware, після чого передає це до uvicorn.run. Метод uvicorn.run надсилає дані запиту, викликаючи метод call Starlette.
Давайте спершу проаналізуємо ініціалізацію Starlette:
class Starlette:
def __init__(
self,
debug: bool = False,
routes: typing.Sequence[BaseRoute] = None,
middleware: typing.Sequence[Middleware] = None,
exception_handlers: typing.Dict[
typing.Union[int, typing.Type[Exception]], typing.Callable
] = None,
on_startup: typing.Sequence[typing.Callable] = None,
on_shutdown: typing.Sequence[typing.Callable] = None,
lifespan: typing.Callable[["Starlette"], typing.AsyncGenerator] = None,
) -> None:
"""
:param debug: Визначає, чи активувати режим налагодження.
:param route: Список маршрутів, які надають HTTP та WebSocket послуги.
:param middleware: Список middleware, які застосовуються до кожного запиту.
:param exception_handler: Словник з обробниками винятків, де HTTP статуси є ключами, а функції зворотного виклику — значеннями.
:on_startup: Функції зворотного виклику, які викликаються під час старту.
:on_shutdown: Функції зворотного виклику, які викликаються під час завершення роботи.
:lifespan: Функція життєвого циклу в ASGI.
"""
# Якщо передано lifespan, то on_startup і on_shutdown не можуть бути передані
# Тому що Starlette фактично перетворює on_start_up і on_shutdown на lifespan для викликів Uvicorn
assert lifespan is None or (
on_startup is None and on_shutdown is None
), "Використовуйте або 'lifespan', або 'on_startup'/'on_shutdown', але не обидва."
# Ініціалізація змінних
self._debug = debug
self.state = State()
self.router = Router(
routes, on_startup=on_startup, on_shutdown=on_shutdown, lifespan=lifespan
)
self.exception_handlers = (
{} if exception_handlers is None else dict(exception_handlers)
)
self.user_middleware = [] if middleware is None else list(middleware)
# Формування стеку middleware
self.middleware_stack = self.build_middleware_stack()
Як видно з коду, ініціалізація вже відповідає більшості вимог.
Однак існує функція для побудови middleware, яка потребує додаткового аналізу:
class Starlette:
def build_middleware_stack(self) -> ASGIApp:
debug = self.debug
error_handler = None
exception_handlers = {}
# Обробка зворотних викликів для винятків та збереження їх в error_handler і exception_handlers
# Лише HTTP статус 500 буде збережений в error_handler
for key, value in self.exception_handlers.items():
if key in (500, Exception):
error_handler = value
else:
exception_handlers[key] = value
# Визначення порядку різних типів middleware
# Перший рівень — ServerErrorMiddleware, який може надрукувати стек помилок, коли буде знайдений виняток, або відобразити сторінку помилки в режимі налагодження для легкого виправлення помилок.
# Другий рівень — це рівень middleware користувача, де буде зберігатися весь middleware, зареєстрований користувачем.
# Третій рівень — це ExceptionMiddleware, який відповідає за обробку винятків та буде обробляти всі винятки, що виникають під час виконання маршруту.
middleware = (
[Middleware(ServerErrorMiddleware, handler=error_handler, debug=debug)]
+ self.user_middleware
+ [
Middleware(
ExceptionMiddleware, handlers=exception_handlers, debug=debug
)
]
)
# В кінці завантажуємо middleware в додаток
app = self.router
for cls, options in reversed(middleware):
# cls — це сама клас middleware, а options — це параметри, які ми передаємо
# Можна побачити, що middleware також є ASGI APP, і завантаження middleware схоже на вкладення одного ASGI APP всередину іншого, як матрьошка.
app = cls(app=app, **options)
# Оскільки middleware завантажується в вкладеному вигляді, а виклик здійснюється через `call_next` для виклику верхнього ASGI APP, використовується зворотний порядок методів.
return app
Після побудови middleware ініціалізація завершена, і метод uvicorn.run викликає метод call:
class Starlette:
async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
scope["app"] = self
await self.middleware_stack(scope, receive, send)
Цей метод простий. Він встановлює додаток у потік запиту через scope для подальших викликів, а потім починає обробку запиту, викликаючи middleware_stack. З цього методу та ініціалізації middleware видно, що middleware в Starlette також є ASGI APP (також можна побачити, що маршрут є ASGI APP на дні стеку викликів). Одночасно Starlette передає обробку винятків до middleware, що рідко зустрічається в інших фреймворках веб-додатків. Можна побачити, що Starlette розроблений так, щоб кожен компонент був максимально можливим ASGI APP.
2. Middleware
Як було сказано раніше, в Starlette middleware є ASGI APP. Тому весь middleware у Starlette повинен бути класом, який відповідає наступному вигляду:
class BaseMiddleware:
def __init__(self, app: ASGIApp) -> None:
pass
async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
pass
У starlette.middleware є багато реалізацій middleware, які відповідають цій вимозі. Однак у цій главі не буде охоплено весь middleware, а лише вибрані кілька представників для аналізу, починаючи з найближчого до маршруту і до найвіддаленішого.
2.1. Middleware для обробки винятків — ExceptionMiddleware
Першим є ExceptionMiddleware. Користувачі не взаємодіють безпосередньо з цим middleware (тому він не розміщений у starlette.middleware), а взаємодіють з ним непрямо через наступний метод:
@app.app_exception_handlers(404)
def not_found_route() -> None: pass
Коли користувач використовує цей метод, Starlette додасть функцію зворотного виклику до відповідного словника, де HTTP статус-код буде ключем, а функція зворотного виклику — значенням.
Коли ExceptionMiddleware виявляє, що під час обробки запиту маршруту виник виняток, він може знайти відповідну функцію зворотного виклику через HTTP статус-код виняткового відповіді, передати запит та виняток до користувацької функції зворотного виклику, і, зрештою, повернути результат функції зворотного виклику користувача назад до попереднього ASGI APP.
Крім того, ExceptionMiddleware також підтримує реєстрацію винятків. Коли виняток, що виникає на маршруті, співпадає з зареєстрованим винятком, викликається відповідна функція зворотного виклику для цього реєстрації винятку.
Ось вихідний код і коментарі до цього класу:
class ExceptionMiddleware:
def __init__(
self, app: ASGIApp, handlers: dict = None, debug: bool = False
) -> None:
self.app = app
self.debug = debug # TODO: Потрібно обробляти випадки 404, якщо увімкнено налагодження.
# Starlette підтримує як HTTP статус-коди, так і типи винятків
self._status_handlers = {} # type: typing.Dict[int, typing.Callable]
self._exception_handlers = {
HTTPException: self.http_exception
} # type: typing.Dict[typing.Type[Exception], typing.Callable]
if handlers is not None:
for key, value in handlers.items():
self.add_exception_handler(key, value)
def add_exception_handler(
self,
exc_class_or_status_code: typing.Union[int, typing.Type[Exception]],
handler: typing.Callable,
) -> None:
# Зворотні виклики винятків, які реєструються користувачем через метод додатку Starlette, в кінці реєструються у _status_handlers або _exception_handler цього класу через цей метод.
if isinstance(exc_class_or_status_code, int):
self._status_handlers[exc_class_or_status_code] = handler
else:
assert issubclass(exc_class_or_status_code, Exception)
self._exception_handlers[exc_class_or_status_code] = handler
def _lookup_exception_handler(
self, exc: Exception
) -> typing.Optional[typing.Callable]:
# Шукає функцію зворотного виклику, пов’язану з зареєстрованим винятком, знаходить відповідну функцію зворотного виклику для винятку через mro
#
# Користувач може зареєструвати базовий клас, і наступні підкласи зареєстрованого винятку також викличуть функцію зворотного виклику, зареєстровану для базового класу.
# Наприклад, користувач реєструє базовий клас, а потім є два винятки, виняток користувача та системний виняток, обидва з яких успадковують цей базовий клас.
# Коли пізніше буде викинуто виняток користувача або системний виняток, буде виконана відповідна функція зворотного виклику, зареєстрована для базового класу.
для cls в type(exc).__mro__:
якщо cls є в self._exception_handlers:
повернути self._exception_handlers[cls]
повернути None
async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
# Знайомий метод виклику ASGI
якщо scope["type"]!= "http":
# Не підтримуються запити WebSocket
await self.app(scope, receive, send)
повернути
# Запобігаємо кільком виняткам в одному відгуку
response_started = False
async def sender(message: Message) -> None:
nonlocal response_started
якщо message["type"] == "http.response.start":
response_started = True
await send(message)
try:
# Викликаємо наступний ASGI APP
await self.app(scope, receive, sender)
except Exception as exc:
handler = None
якщо isinstance(exc, HTTPException):
# Якщо це HTTPException, шукаємо його в реєстрі HTTP зворотних викликів
handler = self._status_handlers.get(exc.status_code)
якщо handler is None:
# Якщо це звичайний виняток, шукаємо його в реєстрі зворотних викликів для винятків
handler = self._lookup_exception_handler(exc)
якщо handler is None:
# Якщо відповідний виняток не знайдений, піднімемо його вище
raise exc from None
# Обробляємо лише один виняток на відповідь
якщо response_started:
msg = "Зловлено оброблений виняток, але відповідь вже почалась."
raise RuntimeError(msg) from exc
request = Request(scope, receive=receive)
якщо asyncio.iscoroutinefunction(handler):
response = await handler(request, exc)
else:
response = await run_in_threadpool(handler, request, exc)
# Обробляємо запит з відповіддю, згенерованою функцією зворотного виклику
await response(scope, receive, sender)
2.2. Користувацький Middleware
Далі йде користувацький middleware, з яким ми найчастіше стикаємось. При використанні starlette.middleware ми зазвичай успадковуємо від middleware, що називається BaseHTTPMiddleware, і розширюємо його на основі наступного коду:
class DemoMiddleware(BaseHTTPMiddleware):
def __init__(
self,
app: ASGIApp,
) -> None:
super(DemoMiddleware, self).__init__(app)
async def dispatch(self, request: Request, call_next: RequestResponseEndpoint) -> Response:
# До
response: Response = await call_next(request)
# Після
return response
Якщо ви хочете виконати попередню обробку перед запитом, напишіть відповідний код у блоці before. Якщо ви хочете обробляти після запиту, пишіть код у блоці after. Використовувати це дуже просто, і вони знаходяться в одному просторі, що означає, що змінні в цьому методі не потребують поширення через контекст чи динамічні змінні (якщо ви працювали з реалізацією middleware в Django або Flask, то зрозумієте елегантність реалізації Starlette).
Тепер давайте подивимося, як це реалізовано. Код дуже простий, близько 60 рядків, але з великою кількістю коментарів:
class BaseHTTPMiddleware:
def __init__(self, app: ASGIApp, dispatch: DispatchFunction = None) -> None:
# Призначаємо наступний рівень ASGI додатку
self.app = app
# Якщо користувач передає dispatch, використовуємо передану функцію, в іншому випадку використовуємо свою функцію dispatch
# Зазвичай користувачі успадковують від BaseHTTPMiddleware і перевизначають метод dispatch
self.dispatch_func = self.dispatch if dispatch is None else dispatch
async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
"""
Функція з підписом стандартної функції ASGI, що вказує на те, що ASGI запити будуть входити через цей метод.
"""
якщо scope["type"]!= "http":
# Якщо тип не http, middleware не буде передано (тобто WebSocket не підтримується)
# Для підтримки WebSocket, реалізувати middleware таким чином дуже складно. Коли я реалізував фреймворк rap, я пожертвував деякими функціями, щоб досягти обробки middleware для трафіку, схожого на WebSocket.
await self.app(scope, receive, send)
return
# Генерація об'єкта запиту з scope
request = Request(scope, receive=receive)
# Вхід в логіку обробки, тобто обробка користувачем
# Відповідь, отримана з цієї логіки, фактично генерується функцією call_next, а функція dispatch лише передає виклик.
response = await self.dispatch_func(request, self.call_next)
# Повертаємо дані до верхнього рівня згідно з отриманою відповіддю
await response(scope, receive, send)
async def call_next(self, request: Request) -> Response:
loop = asyncio.get_event_loop()
# Отримуємо повідомлення наступного рівня через модель виробництва та споживання черги
queue: "asyncio.Queue[typing.Optional[Message]]" = asyncio.Queue()
scope = request.scope
# Передаємо об'єкт receive від uvicorn через request.receive
# Об'єкт receive, що використовується тут, все ще ініціалізований uvicorn
receive = request.receive
send = queue.put
async def coro() -> None:
try:
await self.app(scope, receive, send)
finally:
# Операція put гарантує, що сторона get не буде заблокована
await queue.put(None)
# Запускаємо наступний ASGI APP в іншій корутині через loop.create_task
task = loop.create_task(coro())
# Чекаємо на повернення наступного ASGI APP
message = await queue.get()
if message is None:
# Якщо отримане значення порожнє, це означає, що наступний ASGI APP не повернув відповідь, і могла статися помилка.
# Викликавши task.result(), якщо у корутині сталася помилка, буде викинуто помилку корутини.
task.result()
# Якщо помилки не виникло, це може бути через помилку користувача, наприклад, повернення порожньої відповіді.
# У такому випадку неможливо повернути відповідь клієнту, тому необхідно створити виняток, щоб згенерувати відповідь 500.
raise RuntimeError("No response returned.")
# Коли ASGI обробляє відповідь, це відбувається в кількох етапах. Зазвичай, вище зазначений queue.get є першим етапом отримання відповіді.
assert message["type"] == "http.response.start"
async def body_stream() -> typing.AsyncGenerator[bytes, None]:
# Інша обробка буде передана функції body_stream
# Цей метод просто продовжує повертати потік даних
while True:
message = await queue.get()
if message is None:
break
assert message["type"] == "http.response.body"
yield message.get("body", b"")
task.result()
# Вставляємо функцію body_stream в метод Response
# Відповідь сама є класом, подібним до ASGI APP. Згідно
2.3. ServerErrorMiddleware
ServerErrorMiddleware дуже схожий на ExceptionMiddleware (тому цю частину не будемо детально описувати). Їхня загальна логіка здебільшого однакова. Однак, поки ExceptionMiddleware відповідає за захоплення та обробку винятків маршруту, ServerErrorMiddleware в основному є резервним заходом, щоб завжди забезпечити повернення дійсної HTTP відповіді.
Непрямий виклик функції ServerErrorMiddleware такий самий, як і у ExceptionMiddleware. Але лише коли зареєстрований HTTP статус-код є 500, буде зареєстровано функцію зворотного виклику в ServerErrorMiddleware:
@app.exception_handlers(500)
def not_found_route() -> None: pass
ServerErrorMiddleware знаходиться на найвищому рівні ASGI APP. Він бере на себе завдання обробки резервних винятків. Його завдання просте: якщо під час обробки наступного ASGI APP виникне виняток, він увійде в резервну логіку:
- 1. Якщо увімкнено налагодження, повернути сторінку налагодження.
- 2. Якщо є зареєстрований зворотний виклик, виконати цей зворотний виклик.
- 3. Якщо нічого з вищеописаного немає, повернути відповідь 500.
3. Маршрути
У Starlette маршрути поділяються на дві частини. Одна частина, яку я називаю Router реального додатку, знаходиться на рівні нижче middleware.
Він відповідає майже за все в Starlette, окрім middleware, головним чином це пошук і співставлення маршрутів, обробка запуску та завершення додатку тощо. Інша частина складається з маршрутів, зареєстрованих в Router.
3.1. Router
Router простий. Його основна відповідальність — завантаження та співставлення маршрутів. Ось вихідний код і коментарі, без частини завантаження маршрутів:
class Router:
def __init__(
self,
routes: typing.Sequence[BaseRoute] = None,
redirect_slashes: bool = True,
default: ASGIApp = None,
on_startup: typing.Sequence[typing.Callable] = None,
on_shutdown: typing.Sequence[typing.Callable] = None,
lifespan: typing.Callable[[typing.Any], typing.AsyncGenerator] = None,
) -> None:
# Завантажуємо інформацію з ініціалізації Starlette
self.routes = [] if routes is None else list(routes)
self.redirect_slashes = redirect_slashes
self.default = self.not_found if default is None else default
self.on_startup = [] if on_startup is None else list(on_startup)
self.on_shutdown = [] if on_shutdown is None else list(on_shutdown)
async def default_lifespan(app: typing.Any) -> typing.AsyncGenerator:
await self.startup()
yield
await self.shutdown()
# Якщо ініціалізований lifespan порожній, перетворюємо on_startup і on_shutdown на lifespan
self.lifespan_context = default_lifespan if lifespan is None else lifespan
async def not_found(self, scope: Scope, receive: Receive, send: Send) -> None:
"""Логіка, яка виконується, коли маршрут не знайдено"""
if scope["type"] == "websocket":
# Не вдалося співставити WebSocket
websocket_close = WebSocketClose()
await websocket_close(scope, receive, send)
return
# Якщо ми працюємо в додатку starlette, то піднімемо
# виняток, щоб налаштовуваний обробник винятків міг обробити
# повернення відповіді. Для простих ASGI додатків просто повернемо відповідь.
if "app" in scope:
# В методі __call__ додатку starlette.applications ми можемо побачити, що starlette зберігає себе в scope.
# Після підняття винятку тут, він може бути спійманий ServerErrorMiddleware.
raise HTTPException(status_code=404)
else:
# Для викликів не з Starlette, безпосередньо повертаємо помилку
response = PlainTextResponse("Not Found", status_code=404)
await response(scope, receive, send)
async def lifespan(self, scope: Scope, receive: Receive, send: Send) -> None:
"""
Обробка повідомлень ASGI для lifespan, що дозволяє нам управляти подіями запуску та завершення додатку.
"""
# Логіка виконання lifespan. Під час виконання Starlette взаємодіє з ASGI сервером. Однак наразі можуть бути деякі функції, які ще не розроблені в цьому коді.
first = True
app = scope.get("app")
await receive()
try:
if inspect.isasyncgenfunction(self.lifespan_context):
async for item in self.lifespan_context(app):
assert first, "Lifespan context yielded multiple times."
first = False
await send({"type": "lifespan.startup.complete"})
await receive()
else:
for item in self.lifespan_context(app): # type: ignore
assert first, "Lifespan context yielded multiple times."
first = False
await send({"type": "lifespan.startup.complete"})
await receive()
except BaseException:
if first:
exc_text = traceback.format_exc()
await send({"type": "lifespan.startup.failed", "message": exc_text})
raise
else:
await send({"type": "lifespan.shutdown.complete"})
async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
"""
Головна точка входу для класу Router.
"""
"""
# Головна функція для співставлення та виконання маршрутів
# Наразі підтримуються лише типи http, websocket, lifespan
assert scope["type"] in ("http", "websocket", "lifespan")
# Ініціалізація маршрутизатора в scope
if "router" not in scope:
scope["router"] = self
if scope["type"] == "lifespan":
# Виконання логіки lifespan
await self.lifespan(scope, receive, send)
return
partial = None
# Виконання співставлення маршрутів
for route in self.routes:
match, child_scope = route.matches(scope)
if match == Match.FULL:
# Якщо це повне співпадіння (як URL, так і метод співпадають)
# Тоді виконується звичайна обробка маршруту
scope.update(child_scope)
await route.handle(scope, receive, send)
return
elif match == Match.PARTIAL and partial is None:
# Якщо це часткове співпадіння (URL співпадає, метод не співпадає)
# Тоді зберігаємо значення і продовжуємо співставлення
partial = route
partial_scope = child_scope
if partial is not None:
# Якщо є маршрут з частковим співпадінням, також продовжуємо виконання, але маршрут відповість з помилкою HTTP методу
scope.update(partial_scope)
await partial.handle(scope, receive, send)
return
if scope["type"] == "http" and self.redirect_slashes and scope["path"]!= "/":
# Якщо співпадіння не відбулося, визначаємо редирекцію
redirect_scope = dict(scope)
if scope["path"].endswith("/"):
redirect_scope["path"] = redirect_scope["path"].rstrip("/")
else:
redirect_scope["path"] = redirect_scope["path"] + "/"
for route in self.routes:
match, child_scope = route.matches(redirect_scope)
if match!= Match.NONE:
# Співпадіння знову. Якщо результат не порожній, відправляємо відповідь редирекції
redirect_url = URL(scope=redirect_scope)
response = RedirectResponse(url=str(redirect_url))
await response(scope, receive, send)
return
# Якщо жоден з вищеописаних процесів не спрацював, це означає, що маршрут не знайдений. Тоді буде виконано маршрут за замовчуванням, і відповідь за замовчуванням буде 404 не знайдено
await self.default(scope, receive, send)
Як видно, код Router досить простий. Більшість коду зосереджено в методі call. Однак є кілька проходів для запиту маршрутів, і кожен маршрут виконує регулярний вираз для визначення, чи співпадає він. Дехто може подумати, що швидкість виконання цього процесу є повільною. Я теж так думав раніше, а потім реалізував дерево маршрутів для заміни цього (див. route_trie.py для деталей). Але після тестування продуктивності я виявив, що коли кількість маршрутів не перевищує 50, продуктивність співставлення через цикл краща, ніж через дерево маршрутів. Коли їх кількість не перевищує 100, обидва варіанти рівнозначні. І зазвичай кількість маршрутів, які ми вказуємо, не перевищує 100. Тому не варто хвилюватися про продуктивність співставлення цієї частини маршрутів. Якщо ви все ще переживаєте, можна використовувати Mount, щоб групувати маршрути та зменшити кількість співставлень.
3.2. Інші маршрути
Mount успадковує від BaseRoute, і так само інші маршрути, як-от HostRoute, WebSocketRoute. Вони надають подібні методи, з невеликими відмінностями в реалізації (головним чином у ініціалізації, співставленні маршрутів і зворотному пошуку). Спершу давайте подивимося на BaseRoute:
class BaseRoute:
def matches(self, scope: Scope) -> typing.Tuple[Match, Scope]:
# Стандартний підпис функції для співставлення. Кожен маршрут має повертати кортеж (Match, Scope).
# Match має 3 типи:
# NONE: Немає співпадіння.
# PARTIAL: Часткове співпадіння (URL співпадає, метод не співпадає).
# FULL: Повне співпадіння (як URL, так і метод співпадають).
"""
# Головна функція для співставлення та виконання маршрутів
# Наразі підтримуються лише типи http, websocket, lifespan
assert scope["type"] in ("http", "websocket", "lifespan")
# Ініціалізація маршрутизатора в scope
if "router" not in scope:
scope["router"] = self
if scope["type"] == "lifespan":
# Виконання логіки lifespan
await self.lifespan(scope, receive, send)
return
partial = None
# Виконання співставлення маршрутів
for route in self.routes:
match, child_scope = route.matches(scope)
if match == Match.FULL:
# Якщо це повне співпадіння (як URL, так і метод співпадають)
# Тоді виконується звичайна обробка маршруту
scope.update(child_scope)
await route.handle(scope, receive, send)
return
elif match == Match.PARTIAL and partial is None:
# Якщо це часткове співпадіння (URL співпадає, метод не співпадає)
# Тоді зберігаємо значення і продовжуємо співставлення
partial = route
partial_scope = child_scope
if partial is not None:
# Якщо є маршрут з частковим співпадінням, також продовжуємо виконання, але маршрут відповість з помилкою HTTP методу
scope.update(partial_scope)
await partial.handle(scope, receive, send)
return
if scope["type"] == "http" and self.redirect_slashes and scope["path"]!= "/":
# Якщо співпадіння не відбулося, визначаємо редирекцію
redirect_scope = dict(scope)
if scope["path"].endswith("/"):
redirect_scope["path"] = redirect_scope["path"].rstrip("/")
else:
redirect_scope["path"] = redirect_scope["path"] + "/"
for route in self.routes:
match, child_scope = route.matches(redirect_scope)
if match!= Match.NONE:
# Співпадіння знову. Якщо результат не порожній, відправляємо відповідь редирекції
redirect_url = URL(scope=redirect_scope)
response = RedirectResponse(url=str(redirect_url))
await response(scope, receive, send)
return
# Якщо жоден з вищеописаних процесів не спрацював, це означає, що маршрут не знайдений. Тоді буде виконано маршрут за замовчуванням, і відповідь за замовчуванням буде 404 не знайдено
await self.default(scope, receive, send)
Як видно, код Router досить простий. Більшість коду зосереджено в методі call. Однак є кілька проходів для запиту маршрутів, і кожен маршрут виконує регулярний вираз для визначення, чи співпадає він. Дехто може подумати, що швидкість виконання цього процесу є повільною. Я теж так думав раніше, а потім реалізував дерево маршрутів для заміни цього (див. route_trie.py для деталей). Але після тестування продуктивності я виявив, що коли кількість маршрутів не перевищує 50, продуктивність співставлення через цикл краща, ніж через дерево маршрутів. Коли їх кількість не перевищує 100, обидва варіанти рівнозначні. І зазвичай кількість маршрутів, які ми вказуємо, не перевищує 100. Тому не варто хвилюватися про продуктивність співставлення цієї частини маршрутів. Якщо ви все ще переживаєте, можна використовувати Mount, щоб групувати маршрути та зменшити кількість співставлень.
3.2. Інші маршрути
Mount успадковує від BaseRoute, і так само інші маршрути, як-от HostRoute, WebSocketRoute. Вони надають подібні методи, з невеликими відмінностями в реалізації (головним чином у ініціалізації, співставленні маршрутів і зворотному пошуку). Спершу давайте подивимося на BaseRoute:
class BaseRoute:
def matches(self, scope: Scope) -> typing.Tuple[Match, Scope]:
# Стандартний підпис функції для співпадання. Кожен маршрут має повертати кортеж (Match, Scope).
# Match має 3 типи:
# NONE: Немає співпадіння.
# PARTIAL: Часткове співпадіння (URL співпадає, метод не співпадає).
# FULL: Повне співпадіння (як URL, так і метод співпадають).
├── graphql.py # Відповідає за обробку запитів GraphQL
├── __init__.py
├── py.typed # Типові підказки, необхідні для Starlette
├── requests.py # Запити, для отримання даних користувачем
├── responses.py # Відповіді, відповідають за ініціалізацію заголовків та cookies, генерування даних відповіді згідно з різними класами Response, а потім мають інтерфейс виклику класу ASGI. Цей інтерфейс відправляє протокол ASGI до сервісу Uvicorn. Після відправки, якщо є фонові задачі, вони будуть виконані до завершення.
├── routing.py # Маршрутизація
├── schemas.py # Схеми, пов'язані з OpenApi
├── staticfiles.py # Статичні файли
├── status.py # HTTP статус-коди
├── templating.py # Тemplates для відповідей на основі Jinja
├── testclient.py # Тестовий клієнт
├── types.py # Типи
└── websockets.py # WebSocket
Є багато файлів вище, і деякі простіші будуть пропущені.
5. Підсумок
На цей момент ми проаналізували кілька важливих функціональних кодів Starlette. Starlette — це відмінна бібліотека з чудовою концепцією дизайну. Рекомендується самостійно прочитати вихідний код Starlette, що допоможе вам при написанні власних фреймворків у майбутньому.
Leapcell: найкраща платформа без сервера для хостингу Python
Нарешті, я хочу поділитися найкращою платформою для розгортання FastAPI: Leapcell
1. Підтримка кількох мов
- Розробка на JavaScript, Python, Go або Rust.
2. Безлімітне розгортання проектів безкоштовно
- Платіть лише за використання — без запитів, без витрат.
3. Неймовірна ефективність витрат
- Платіть за використання — без плат за простоювання.
- Приклад: $25 підтримує 6,94 млн запитів із середнім часом відповіді 60 мс.
4. Простота для розробників
- Інтуїтивно зрозумілий інтерфейс для легкого налаштування.
- Повністю автоматизовані CI/CD пайплайни та інтеграція GitOps.
- Реальний моніторинг метрик і логів для корисних висновків.
5. Легке масштабування та висока продуктивність
- Автоматичне масштабування для обробки високої кількості запитів з легкістю.
- Без операційних витрат — зосередьтеся на створенні.
Twitter Leapcell: https://x.com/LeapcellHQ
Перекладено з: The Core of FastAPI: A Deep Dive into Starlette 🌟🌟🌟