-
Notifications
You must be signed in to change notification settings - Fork 0
/
Copy pathapp.py
85 lines (70 loc) · 2.43 KB
/
app.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
from starlette.applications import Starlette
from starlette.routing import Route
from starlette.requests import HTTPConnection, Request
from starlette.responses import Response
from starlette.types import Message
from starlette.middleware import Middleware
from starlette_context.middleware import RawContextMiddleware
from starlette_context.plugins import Plugin
from starlette_context import context
from aiohttp import ClientSession, ClientTimeout
from shikithon import ShikimoriAPI
from jinja2 import Environment, FileSystemLoader
from views.main import index, migrate_route
from views.cards import user_card, collection_card, bingo_card
from typing import Any, Optional, Union
class AiohttpSessionPlugin(Plugin):
key = "aiohttp"
async def process_request(
self, _: Union[Request, HTTPConnection]
) -> Optional[Any]:
"""Runs always on request."""
return ClientSession(
timeout=ClientTimeout(10),
trust_env=True
)
async def enrich_response(self, _: Union[Response, Message]) -> None:
"""Runs always on response."""
session: ClientSession = context.get("aiohttp")
await session.close()
class ShikimoriSessionPlugin(Plugin):
key = "shikimori"
async def process_request(
self, _: Union[Request, HTTPConnection]
) -> Optional[Any]:
"""Runs always on request."""
return await ShikimoriAPI().open()
async def enrich_response(self, _: Union[Response, Message]) -> None:
"""Runs always on response."""
session: ShikimoriAPI = context.get("shikimori")
await session.close()
class JinjaPlugin(Plugin):
key = "jinja"
async def process_request(
self, _: Union[Request, HTTPConnection]
) -> Optional[Any]:
"""Runs always on request."""
return Environment(
trim_blocks=True,
loader=FileSystemLoader("src/cards/"),
auto_reload=False
)
app = Starlette(
routes=[
Route("/", index),
Route("/migrate", migrate_route),
Route("/user/{user_id:str}", user_card),
Route("/collection/{collection_id:int}", collection_card),
Route("/bingo/{user_id:int}", bingo_card)
],
middleware=[
Middleware(
RawContextMiddleware,
plugins=(
AiohttpSessionPlugin(),
ShikimoriSessionPlugin(),
JinjaPlugin()
)
)
]
)