Skip to content

Instantly share code, notes, and snippets.

@WilliamStam
Created November 15, 2023 08:30
Show Gist options
  • Save WilliamStam/f9c6868da23f7a3dab5101665f097564 to your computer and use it in GitHub Desktop.
Save WilliamStam/f9c6868da23f7a3dab5101665f097564 to your computer and use it in GitHub Desktop.
Litestar gist to prove i was wrong and litestar is right
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