Created
November 15, 2023 08:30
-
-
Save WilliamStam/f9c6868da23f7a3dab5101665f097564 to your computer and use it in GitHub Desktop.
Litestar gist to prove i was wrong and litestar is right
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
import logging | |
from pathlib import Path | |
from litestar import Controller, Litestar, Request, Router, get | |
from litestar.connection import ASGIConnection | |
from litestar.exceptions import PermissionDeniedException | |
from litestar.handlers import BaseRouteHandler | |
from litestar.logging import LoggingConfig | |
from litestar.openapi import OpenAPIConfig | |
from litestar.openapi.spec import Components, SecurityScheme | |
from litestar import Request | |
from litestar.types import ASGIApp, Receive, Scope, Send | |
log_config = LoggingConfig( | |
root={ | |
"level": logging.getLevelName(logging.INFO), | |
"handlers": [ | |
"console" | |
] | |
}, | |
formatters={ | |
"standard": { | |
"format": "%(asctime)s %(process)s %(levelname)s:%(name)s: %(message)s" | |
} | |
}, | |
) | |
logger = log_config.configure()() | |
def MyMiddleware(app: ASGIApp) -> ASGIApp: | |
async def middleware(scope: Scope, receive: Receive, send: Send) -> None: | |
request = Request(scope) | |
request.logger.info("MyMiddleware") | |
scope['in_middleware'] = True | |
await app(scope, receive, send) | |
return middleware | |
class MyGuard(): | |
def __init__(self): | |
pass | |
async def __call__(self, connection: ASGIConnection, route_handler: BaseRouteHandler): | |
middleware_ran = connection.scope.get('in_middleware',False) | |
connection.logger.info(f"MyGuard") | |
connection.logger.info(f"MyGuard checking middleware ran: {middleware_ran}") | |
connection.scope['in_guard'] = True | |
class MyController1(Controller): | |
@get("/MyController1") | |
async def MyController1(self) -> str: | |
return f""" | |
INFO:root: MyMiddleware | |
INFO: 127.0.0.1:63435 - "GET /MyController1 HTTP/1.1" 200 OK | |
""" | |
@get("/MyController1_route_guard", guards=[MyGuard()]) | |
async def MyController1_route_guard(self,request: Request) -> str: | |
middleware_ran = request.scope.get('in_middleware', False) | |
guard_ran = request.scope.get('in_guard', False) | |
request.logger.info((f"MyController1.with_guard checking | middleware: {middleware_ran} | guard: {guard_ran}")) | |
return f""" | |
INFO:root: MyMiddleware | |
INFO:root: MyGuard | |
INFO:root: MyGuard checking middleware ran: True | |
INFO:root: MyController1.with_guard checking | middleware: True | guard: True | |
INFO: 127.0.0.1:63664 - "GET /MyController1_route_guard HTTP/1.1" 200 OK | |
""" | |
class MyController2(Controller): | |
@get("/MyController2_router_guard") | |
async def MyController2_router_guard(self) -> str: | |
return """ | |
INFO:root: MyMiddleware | |
INFO:root: MyGuard | |
INFO:root: MyGuard checking middleware ran: True | |
INFO: 127.0.0.1:63757 - "GET /router_guard_check/MyController2_router_guard HTTP/1.1" 200 OK | |
""" | |
@get("/MyController2_router_guard_route_guard", guards=[MyGuard()]) | |
async def MyController2_router_guard_route_guard(self, request: Request) -> str: | |
middleware_ran = request.scope.get('in_middleware', False) | |
guard_ran = request.scope.get('in_guard', False) | |
request.logger.info((f"MyController2.with_guard checking | middleware: {middleware_ran} | guard: {guard_ran}")) | |
return """ | |
INFO:root: MyMiddleware | |
INFO:root: MyGuard | |
INFO:root: MyGuard checking middleware ran: True | |
INFO:root: MyGuard | |
INFO:root: MyGuard checking middleware ran: True | |
INFO:root: MyController2.with_guard checking | middleware: True | guard: True | |
INFO: 127.0.0.1:63781 - "GET /router_guard_check/MyController2_router_guard_route_guard HTTP/1.1" 200 OK | |
""" | |
router_guard_check = Router( | |
"/", | |
route_handlers=[ | |
MyController2, | |
], | |
guards=[MyGuard()] | |
) | |
class MyGuardWithException(): | |
def __init__(self,trigger=False): | |
self.trigger = trigger | |
async def __call__(self, connection: ASGIConnection, route_handler: BaseRouteHandler): | |
middleware_ran = connection.scope.get('in_middleware', False) | |
connection.logger.info(f"MyGuardWithException - triggered {self.trigger}") | |
connection.logger.info(f"MyGuardWithException checking middleware ran: {middleware_ran}") | |
connection.scope['in_guard'] = True | |
if self.trigger: | |
raise PermissionDeniedException() | |
class MyController3(Controller): | |
@get("/MyController3_router_guard_with_no_exception") | |
async def MyController3_router_guard_with_no_exception(self) -> str: | |
return f""" | |
INFO:root: MyMiddleware | |
INFO:root: MyGuardWithException - triggered False | |
INFO:root: MyGuardWithException checking middleware ran: True | |
INFO: 127.0.0.1:63947 - "GET /MyController3_router_guard_with_no_exception HTTP/1.1" 200 OK | |
""" | |
@get("/MyController3_router_guard_with_exception_route_guard_with_exception_triggered", guards=[MyGuardWithException(True)]) | |
async def MyController3_router_guard_with_exception_route_guard_with_exception_triggered(self, request: Request) -> str: | |
middleware_ran = request.scope.get('in_middleware', False) | |
guard_ran = request.scope.get('in_guard', False) | |
request.logger.info((f"MyController3.with_guard checking | middleware: {middleware_ran} | guard: {guard_ran}")) | |
return """ | |
INFO:root: MyMiddleware | |
INFO:root: MyGuardWithException - triggered False | |
INFO:root: MyGuardWithException checking middleware ran: True | |
INFO:root: MyGuardWithException - triggered True | |
INFO:root: MyGuardWithException checking middleware ran: True | |
INFO: 127.0.0.1:64085 - "GET /MyController3_router_guard_with_exception_route_guard_with_exception_triggered HTTP/1.1" 403 Forbidden | |
""" | |
class MyController4(Controller): | |
@get("/MyController4_router_guard_exception_triggered_route_guard_exception") | |
async def MyController4_router_guard_exception_triggered_route_guard_exception(self) -> str: | |
return f""" | |
INFO:root: MyMiddleware | |
INFO:root: MyGuardWithException - triggered True | |
INFO:root: MyGuardWithException checking middleware ran: True | |
INFO: 127.0.0.1:64167 - "GET /MyController4_router_guard_exception_triggered_route_guard_exception HTTP/1.1" 403 Forbidden | |
""" | |
@get("/MyController4_router_guard_exception_triggered_route_guard_exception_triggered", guards=[MyGuardWithException(True)]) | |
async def MyController4_router_guard_exception_triggered_route_guard_exception_triggered(self, request: Request) -> str: | |
middleware_ran = request.scope.get('in_middleware', False) | |
guard_ran = request.scope.get('in_guard', False) | |
request.logger.info((f"MyController4.with_guard checking | middleware: {middleware_ran} | guard: {guard_ran}")) | |
return """ | |
INFO:root: MyMiddleware | |
INFO:root: MyGuardWithException - triggered True | |
INFO:root: MyGuardWithException checking middleware ran: True | |
INFO: 127.0.0.1:64316 - "GET /MyController4_router_guard_exception_triggered_route_guard_exception_triggered HTTP/1.1" 403 Forbidden | |
""" | |
router_guard_check_with_exception_on_route = Router( | |
"/", | |
route_handlers=[ | |
MyController3, | |
], | |
guards=[MyGuardWithException()] | |
) | |
router_guard_check_with_exception_on_router = Router( | |
"/", | |
route_handlers=[ | |
MyController4, | |
], | |
guards=[MyGuardWithException(True)] | |
) | |
async def new_line(request: Request) -> None: | |
request.logger.info("------------------") | |
return None | |
app = Litestar( | |
debug=True, | |
route_handlers=[ | |
MyController1, | |
router_guard_check, | |
router_guard_check_with_exception_on_route, | |
router_guard_check_with_exception_on_router | |
], | |
after_response=new_line, | |
openapi_config=OpenAPIConfig( | |
title="my api", | |
version="1.0.0", | |
security=[], | |
components=Components( | |
security_schemes={ | |
"APIKeyQuery": SecurityScheme( | |
type="apiKey", | |
description="Pass the token via a query sting item ?token=xxx", | |
security_scheme_in="query", | |
name="token", | |
) | |
}, | |
), | |
), | |
middleware=[ | |
MyMiddleware | |
], | |
logging_config=log_config | |
) |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment