Coverage for fastblocks / adapters / app / default.py: 3%
212 statements
« prev ^ index » next coverage.py v7.12.0, created at 2025-11-26 03:58 -0800
1"""Default App Adapter for FastBlocks.
3Provides the main FastBlocks application instance with lifecycle management,
4startup/shutdown sequences, and adapter integration.
6Author: lesleslie <les@wedgwoodwebworks.com>
7Created: 2025-01-12
8"""
10import typing as t
11from base64 import b64encode
12from contextlib import asynccontextmanager, suppress
13from time import perf_counter
14from uuid import UUID
16from acb.adapters import AdapterStatus, get_adapter
17from acb.depends import depends
18from starlette.types import ASGIApp, Receive, Scope, Send
19from fastblocks.applications import FastBlocks
21from ._base import AppBase, AppBaseSettings
23main_start = perf_counter()
24Cache = Storage = None
27class AppSettings(AppBaseSettings):
28 url: str = "http://localhost:8000"
29 token_id: str | None = "_fb_"
31 def __init__(self, **data: t.Any) -> None:
32 super().__init__(**data)
33 # Note: URL configuration moved to runtime initialization
34 # to avoid coroutine access in __init__
35 token_prefix = self.token_id or "_fb_"
36 self.token_id = "".join(
37 [token_prefix, b64encode(self.name.encode()).decode().rstrip("=")],
38 )
41class FastBlocksApp(FastBlocks):
42 def __init__(self, **kwargs: t.Any) -> None:
43 super().__init__(lifespan=self.lifespan, **kwargs)
45 async def init(self) -> None:
46 pass
48 def _get_startup_time(self) -> float:
49 startup_time = getattr(self, "_startup_time", None)
50 if startup_time is None or startup_time <= 0:
51 import time
53 init_start = getattr(self, "_init_start_time", None)
54 startup_time = time.time() - init_start if init_start else 0.001
55 return startup_time
57 def _get_debug_enabled(self, config: t.Any) -> list[str]:
58 debug_enabled = []
59 if hasattr(config, "debug"):
60 for key, value in vars(config.debug).items():
61 if value and key != "production":
62 debug_enabled.append(key)
63 return debug_enabled
65 def _get_color_constants(self) -> dict[str, str]:
66 return {
67 "GREEN": "\033[92m",
68 "BLUE": "\033[94m",
69 "YELLOW": "\033[93m",
70 "CYAN": "\033[96m",
71 "RESET": "\033[0m",
72 "BOLD": "\033[1m",
73 }
75 def _format_info_lines(
76 self,
77 config: t.Any,
78 colors: dict[str, str],
79 debug_enabled: list[str],
80 startup_time: float,
81 ) -> list[str]:
82 app_title = getattr(config.app, "title", "Welcome to FastBlocks")
83 app_domain = getattr(config.app, "domain", "localhost")
84 debug_str = ", ".join(debug_enabled) if debug_enabled else "disabled"
86 return [
87 f"{colors['CYAN']}{colors['BOLD']}{app_title}{colors['RESET']}",
88 f"{colors['BLUE']}Domain: {app_domain}{colors['RESET']}",
89 f"{colors['YELLOW']}Debug: {debug_str}{colors['RESET']}",
90 f"{colors['YELLOW']}══════════════════════════════════════════════════{colors['RESET']}",
91 f"{colors['GREEN']}🚀 FastBlocks Application Ready{colors['RESET']}",
92 f"{colors['YELLOW']}⚡ Startup time: {startup_time * 1000:.2f}ms{colors['RESET']}",
93 f"{colors['CYAN']}🌐 Server running on http://127.0.0.1:8000{colors['RESET']}",
94 f"{colors['YELLOW']}══════════════════════════════════════════════════{colors['RESET']}",
95 ]
97 def _clean_and_center_line(self, line: str, colors: dict[str, str]) -> str:
98 line_clean = line
99 for color in colors.values():
100 line_clean = line_clean.replace(color, "")
101 line_width = len(line_clean)
102 padding = max(0, (90 - line_width) // 2)
103 return " " * padding + line
105 async def _display_fancy_startup(self) -> None:
106 from acb.depends import depends
107 from aioconsole import aprint
108 from pyfiglet import Figlet
110 config = await depends.get("config")
111 app_name = getattr(config.app, "name", "FastBlocks")
112 startup_time = self._get_startup_time()
113 debug_enabled = self._get_debug_enabled(config)
114 colors = self._get_color_constants()
115 banner = Figlet(font="slant", width=90, justify="center").renderText(
116 app_name.upper(),
117 )
118 await aprint(f"\n\n{banner}\n")
119 info_lines = self._format_info_lines(
120 config,
121 colors,
122 debug_enabled,
123 startup_time,
124 )
125 for line in info_lines:
126 self._clean_and_center_line(line, colors)
128 async def _display_simple_startup(self) -> None:
129 from contextlib import suppress
131 with suppress(Exception):
132 from acb.depends import depends
134 config = await depends.get("config")
135 getattr(config.app, "name", "FastBlocks")
136 self._get_startup_time()
138 async def post_startup(self) -> None:
139 try:
140 await self._display_fancy_startup()
141 except Exception:
142 await self._display_simple_startup()
144 @asynccontextmanager
145 async def lifespan(self, app: "FastBlocks") -> t.AsyncIterator[None]:
146 try:
147 logger = getattr(self, "logger", None)
148 if logger:
149 logger.info("FastBlocks application starting up")
150 except Exception as e:
151 logger = getattr(self, "logger", None)
152 if logger:
153 logger.exception(f"Error during startup: {e}")
154 raise
155 yield
156 logger = getattr(self, "logger", None)
157 if logger:
158 logger.info("FastBlocks application shutting down")
161class App(AppBase):
162 settings: AppSettings | None = None
163 router: t.Any = None
164 middleware_manager: t.Any = None
165 templates: t.Any = None
166 models: t.Any = None
167 exception_handlers: t.Any = None
168 middleware_stack: t.Any = None
169 user_middleware: t.Any = None
170 fastblocks_app: t.Any = None
172 def __init__(self, **kwargs: t.Any) -> None:
173 super().__init__(**kwargs)
174 self.settings = AppSettings()
175 self.fastblocks_app = FastBlocksApp()
176 self.router = None
177 self.middleware_manager = None
178 self.templates = None
179 self.models = None
180 self.exception_handlers = {}
181 self.middleware_stack = None
182 self.user_middleware = []
183 self.state = None
185 @property
186 def logger(self) -> t.Any:
187 if hasattr(super(), "logger"):
188 with suppress(Exception):
189 return super().logger
190 try:
191 return depends.get("logger")
192 except Exception:
193 import logging
195 return logging.getLogger(self.__class__.__name__)
197 @logger.setter
198 def logger(self, value: t.Any) -> None:
199 pass
201 @logger.deleter
202 def logger(self) -> None:
203 pass
205 async def init(self) -> None:
206 import time
208 self._init_start_time = time.time()
209 await self.fastblocks_app.init()
210 try:
211 self.templates = await depends.get("templates")
212 except Exception:
213 self.templates = None
214 try:
215 self.models = await depends.get("models")
216 except Exception:
217 self.models = None
218 try:
219 routes_adapter = await depends.get("routes")
220 self.router = routes_adapter
221 self.fastblocks_app.routes.extend(routes_adapter.routes)
222 except Exception:
223 self.router = None
224 self.middleware_manager = None
225 self.exception_handlers = self.fastblocks_app.exception_handlers
226 self.middleware_stack = self.fastblocks_app.middleware_stack
227 self.user_middleware = self.fastblocks_app.user_middleware
228 self.state = self.fastblocks_app.state
229 import time
231 self._startup_time = time.time() - self._init_start_time
232 self.fastblocks_app._startup_time = self._startup_time
233 self.fastblocks_app._init_start_time = self._init_start_time
234 await self.post_startup()
236 def __call__(self, scope: Scope, receive: Receive, send: Send) -> ASGIApp:
237 return t.cast(ASGIApp, self.fastblocks_app(scope, receive, send))
239 def __getattr__(self, name: str) -> t.Any:
240 return getattr(self.fastblocks_app, name)
242 async def post_startup(self) -> None:
243 await self.fastblocks_app.post_startup()
245 async def _setup_admin_adapter(self, app: FastBlocks) -> None:
246 if not get_adapter("admin"):
247 return
248 sql = await depends.get("sql")
249 auth = await depends.get("auth")
250 admin = await depends.get("admin")
251 admin.__init__(
252 app,
253 engine=sql.engine,
254 title=self.config.admin.title,
255 debug=getattr(self.config.debug, "admin", False),
256 base_url=self.config.admin.url,
257 logo_url=self.config.admin.logo_url,
258 authentication_backend=auth,
259 )
260 self.router.routes.insert(0, self.router.routes.pop())
262 async def _startup_sequence(self, app: FastBlocks) -> None:
263 await self._setup_admin_adapter(app)
264 await self.post_startup()
265 main_start_time = perf_counter() - main_start
266 self.logger.warning(f"App started in {main_start_time} s")
268 async def _shutdown_logger(self) -> None:
269 import asyncio
271 completer = None
272 if hasattr(self.logger, "complete"):
273 completer = self.logger.complete()
274 elif hasattr(self.logger, "stop"):
275 completer = self.logger.stop()
276 if completer:
277 await asyncio.wait_for(completer, timeout=1.0)
279 def _cancel_remaining_tasks(self) -> None:
280 import asyncio
282 loop = asyncio.get_event_loop()
283 tasks = [t for t in asyncio.all_tasks(loop) if not t.done()]
284 if tasks:
285 self.logger.debug(f"Cancelling {len(tasks)} remaining tasks")
286 for task in tasks:
287 task.cancel()
289 @asynccontextmanager
290 async def lifespan(self, app: FastBlocks) -> t.AsyncIterator[None]:
291 try:
292 await self._startup_sequence(app)
293 except Exception as e:
294 self.logger.exception(f"Error during startup: {e}")
295 raise
296 yield
297 self.logger.critical("Application shut down")
298 try:
299 await self._shutdown_logger()
300 except TimeoutError:
301 self.logger.warning("Logger completion timed out, forcing shutdown")
302 except Exception as e:
303 self.logger.exception(f"Logger completion failed: {e}")
304 finally:
305 with suppress(Exception):
306 self._cancel_remaining_tasks()
309MODULE_ID = UUID("01937d86-8f6e-7f70-c231-5678901234ef")
310MODULE_STATUS = AdapterStatus.STABLE
312depends.set(App)