Create a routes constants file and a router for getting routes in app

This commit is contained in:
Campbell Alden 2026-08-05 00:12:47 +09:00
parent 790e0a1ce5
commit 5a4f9dc250
5 changed files with 93 additions and 10 deletions

View file

47
src/constants/routes.py Normal file
View file

@ -0,0 +1,47 @@
import abc
from typing import Any, TypedDict
from dataclasses import dataclass, asdict
class ParamCodec[Params]:
@abc.abstractmethod
def parse(self, path) -> Params:
pass
class EmptyParams(ParamCodec[None]):
def parse(self, path) -> None:
pass
@dataclass
class Route[Params: Any]:
path: str
codec: ParamCodec[Params]
def interpolate(self, params: Params) -> str:
interpolated = self.path
if params:
for k, v in asdict(params).items():
interpolated.replace(f':{k}', str(v))
return interpolated
def static(path: str) -> Route[None]:
return Route(path, EmptyParams())
class Routes(TypedDict):
index: Route[None]
confirm_email: Route[None]
signup: Route[None]
dashboard: Route[None]
ROUTES: Routes = {
'index': static('/'),
'confirm_email': static('/confirm-email'),
'signup': static('/register'),
'dashboard': static('/dashboard'),
}

View file

@ -1,5 +1,6 @@
#!/usr/bin/env python #!/usr/bin/env python
from src.services.router import Router
from src.services.notifications import NotificationService, get_notification_service from src.services.notifications import NotificationService, get_notification_service
from src.services.email import EmailService, get_email_service from src.services.email import EmailService, get_email_service
from src.services.auth import AuthService from src.services.auth import AuthService
@ -23,6 +24,7 @@ class AppServices:
auth: AuthService auth: AuthService
email: EmailService email: EmailService
notifications: NotificationService notifications: NotificationService
router: Router
class MyFlask(Flask): class MyFlask(Flask):
@ -31,19 +33,24 @@ class MyFlask(Flask):
def create_app(name: str, config: Config) -> Flask: def create_app(name: str, config: Config) -> Flask:
logging.basicConfig(level=config.logging.level, format=('%(asctime)s %(levelname)s [%(name)s] %(message)s')) logging.basicConfig(level=config.logging.level, format=('%(asctime)s %(levelname)s [%(name)s] %(message)s'))
app = MyFlask(name) app = MyFlask(name)
# Configure Services
database = get_database(config.database) database = get_database(config.database)
user_repo = UserRepoImpl(database) user_repo = UserRepoImpl(database)
email_service = get_email_service(config.email) email_service = get_email_service(config.email)
auth_service = AuthService(config.auth)
router = Router(config.host, config.port)
app.services = AppServices( app.services = AppServices(
users=UserService(user_repo, email_service), users=UserService(repo=user_repo, email_service=email_service, auth_service=auth_service, router=router),
auth=AuthService(config.auth), auth=auth_service,
email=email_service, email=email_service,
notifications=get_notification_service(email_service), notifications=get_notification_service(email_service),
router=router,
) )
if config.host: if config.host != '0.0.0.0':
app.config['SERVER_NAME'] = config.host app.config['SERVER_NAME'] = config.host
return app return app

19
src/services/router.py Normal file
View file

@ -0,0 +1,19 @@
from src.constants.routes import Route
class Router:
def __init__(self, host: str, port: int | None):
self._host = host
self._port = port
def interpolate_url[Params](self, route: Route[Params], params: Params, **query_params) -> str:
query_strs = []
for key, value in query_params.items():
query_strs.append(f'{key}={value}')
query_str = '&'.join(query_strs)
query_str = f'?{query_str}' if len(query_str) > 0 else query_str
return f'{self._host}{"" if self._port is None else f":{self._port}"}{route.interpolate(params)}{query_str}'
def get_url(self, route: Route[None], **query_params):
return self.interpolate_url(route, None, **query_params)

View file

@ -1,3 +1,6 @@
from src.services.router import Router
from typing import TypedDict, Unpack
from src.constants.routes import ROUTES
from src.services.auth import AuthService, EmailConfirmationClaim, JWT from src.services.auth import AuthService, EmailConfirmationClaim, JWT
from email.headerregistry import Address from email.headerregistry import Address
from jinja2 import Environment, PackageLoader, select_autoescape from jinja2 import Environment, PackageLoader, select_autoescape
@ -23,11 +26,19 @@ def is_valid_password(password: str) -> bool:
return long_enough and short_enough and has_symbol return long_enough and short_enough and has_symbol
class ServiceDeps(TypedDict):
repo: UserRepo
email_service: EmailService
auth_service: AuthService
router: Router
class UserService: class UserService:
def __init__(self, repo: UserRepo, email_service: EmailService, auth_service: AuthService): def __init__(self, **kwargs: Unpack[ServiceDeps]):
self._repo = repo self._repo = kwargs['repo']
self._email_service = email_service self._email_service = kwargs['email_service']
self._auth_service = auth_service self._auth_service = kwargs['auth_service']
self._router = kwargs['router']
def login(self, user: UserDTO) -> UserProfile: def login(self, user: UserDTO) -> UserProfile:
full_user = self._repo.auth_as_user(user) full_user = self._repo.auth_as_user(user)
@ -66,8 +77,7 @@ class UserService:
html_template = env.get_template('mail/confirmation_email.html') html_template = env.get_template('mail/confirmation_email.html')
token = self._auth_service.mint_claim_from_user(EmailConfirmationClaim, user) token = self._auth_service.mint_claim_from_user(EmailConfirmationClaim, user)
# TODO: Parameterize this with configuration that also drives the API url = self._router.get_url(ROUTES['confirm_email'], token=token)
url = f'/confirm?token={token}'
text_content = text_template.render(confirmation_link=url) text_content = text_template.render(confirmation_link=url)
html_content = html_template.render(confirmation_link=url) html_content = html_template.render(confirmation_link=url)