diff --git a/docs/adr/041-database-install-command.md b/docs/adr/041-database-install-command.md new file mode 100644 index 00000000..62ef6f90 --- /dev/null +++ b/docs/adr/041-database-install-command.md @@ -0,0 +1,50 @@ +# 041. InstallCommand base class and database:install + +Date: 2026-10-22 + +## Problem + +`fastapi-startkit init --database ...` sets up the database only for new projects. An existing application without a database has to copy a config file and edit `bootstrap/application.py` by hand: import `DatabaseProvider` and its config, and add `(DatabaseProvider, DatabaseConfig)` to `providers=[...]`. `provider:publish` cannot help, because it only knows about the resources of providers that are already registered and booted. + +Other packages (cache, broadcasting, storage, ...) need the same steps: publish a config file and register their provider in `bootstrap/application.py`. If the AST editing lives inside the database command, each package would have to copy it. + +## Alternatives + +1. Register the command from `DatabaseProvider`. It would only exist after the database is installed, which is too late. +2. Generate a provider list in a separate file such as `config/providers.py`. Apps do not use one. Every app built with the installer and every example lists providers inline in the `Application(...)` call in `bootstrap/application.py`. +3. Add a core command, registered by `AppProvider` next to `provider:publish`, that writes the config file and edits `bootstrap/application.py` in place. +4. Option 3, with the bootstrap edit as a module-level function that each command calls directly. This shares the logic, but every command would repeat the `--force` option, the publish-or-skip handling, and the error/manual-steps output. +5. Option 3, with a reusable `InstallCommand` base class. + +## Decision + +Option 5. `fastapi_startkit/console/install_command.py` defines `InstallCommand(Command)` with a fluent class builder: + +```python +CacheInstallCommand = ( + InstallCommand.register("cache:install", "Install the cache.") + .copy(files("my_package").joinpath("stubs/cache.py"), "config/cache.py") + .add_provider(CacheProvider, "config.cache.CacheConfig") +) +``` + +- `register(name, description)`, `copy(stub, destination)` and `add_provider(provider, config=None)` are classmethods. Each one returns a new subclass with the added attributes and leaves the receiver unchanged. The result is a command class, which is what `Provider.commands([...])` expects. +- `add_provider` takes classes or dotted strings. A class is converted to `module.QualName`. A string is needed when the class lives in the user's app (`config.database.DatabaseConfig` does not exist when the framework loads), or when importing it would load an optional extra. Each `ProviderEntry` derives the inserted entry (`(Provider, Config)`, or a bare `Provider` when there is no config) and the `from module import Name` statements. +- `copy` accepts a path or an `importlib.resources` `Traversable` and copies the stub verbatim. +- `handle()` runs `publish()` and then registers every provider. The default `publish()` copies each stub. A subclass that needs options or templating overrides `publish()` and calls `publish_file(relative_path, content)`. `publish_file` creates parent directories and skips an existing file unless `--force` is passed. `--force` is declared on the base class, and subclasses add their own options with `options = [*InstallCommand.options, ...]`. +- `register_provider(source, provider, entry, imports)` is a pure function that returns the edited source, the unchanged source if the provider is already registered, or `None` if there is no `Application(providers=[...])` to edit. + +`DatabaseInstallCommand` is a regular subclass, because it has a `--driver` option and renders the stub template. It declares `providers = (ProviderEntry.of("fastapi_startkit.masoniteorm.DatabaseProvider", "config.database.DatabaseConfig"),)`. `uv run artisan database:install [--driver sqlite|postgres|mysql] [--force]`: + +- Renders `config/database.py` from the installer's `database/config.py.stub`, so it matches what `fastapi-startkit init` generates. `config/` is created if missing. If the file already exists it is skipped with a message, unless `--force` is passed. +- Parses `bootstrap/application.py` with `ast` and finds the `Application(...)` call. If its `providers` list already names `DatabaseProvider`, nothing is changed. Otherwise `(DatabaseProvider, DatabaseConfig)` is appended to the list, following its layout: a new line for multi-line lists, inline for single-line lists. If there is no `providers=` keyword, one is added. Missing `from fastapi_startkit.masoniteorm import DatabaseProvider` and `from config.database import DatabaseConfig` imports are inserted after the last top-level import. +- If `bootstrap/application.py` or the `Application(...)` call is missing, it prints the lines to add by hand and returns 1. The config file is still written. + +The command lives in `fastapi_startkit/console/` rather than `masoniteorm/commands/`, because importing `masoniteorm.commands` loads SQLAlchemy, which is an optional extra. `install_command.py` imports only `ast`, cleo and `Command`, so packages with optional extras can subclass it without loading anything extra. Text is inserted at AST-reported positions rather than regenerated with `ast.unparse`, so the user's formatting and comments are kept. + +## Validation + +- `tests/console/test_install_command.py` tests commands built with the fluent API: a copied stub with a tuple entry, a provider-only command with a bare entry, several providers with a `Traversable` stub, class references converted to imports, and chain calls that do not mutate the base class. It also tests `register_provider` directly on single-line, multi-line, empty and missing `providers` lists, files with no imports, existing imports, already-registered providers and unsupported shapes. +- `tests/console/test_database_install_command.py` runs `database:install` against a temporary app directory. It covers a fresh install, a second run that changes nothing, an existing config with and without `--force`, a missing `config/` directory, a missing bootstrap file, an unsupported driver, and loading the edited bootstrap file. +- The full suite passes with coverage of at least 80%. +- Manual run, twice, against an app generated by `fastapi-startkit init --database none`. diff --git a/docs/adr/AGENTS.md b/docs/adr/AGENTS.md index 7b657eb1..471a18b2 100644 --- a/docs/adr/AGENTS.md +++ b/docs/adr/AGENTS.md @@ -36,3 +36,4 @@ Read this index first, then open the records relevant to the change. Before impl | [032](032-migrate-seed-option.md) | 2026-10-21 | db:migrate:fresh and db:migrate:refresh --seed run seeders in the same event loop | `--seed` called the unregistered `seed:run` through Cleo `call` (which also drops the first argument) inside a running event loop, where `db:seed` would call `asyncio.run` again. The seeder selection moves into async `DBSeedCommand.run_seeders`, which `db:seed`, `db:migrate:fresh` and `db:migrate:refresh` all await. `db:migrate:refresh` is now registered. | | [033](033-async-seeder-stub.md) | 2026-10-21 | Generated seeders have an async run | `create_seed.stub` generated a synchronous `run` with docstrings, but `Seeder` awaits `run()`, so every generated table seeder raised `TypeError`. The stub now generates `async def run(self) -> None` with no docstrings; `Seeder` is unchanged. | | [038](038-storage-binary-copy-and-request-scoped-vite-inertia.md) | 2026-10-25 | Storage binary reads and relative copy; request-scoped Vite and Inertia state | `LocalDriver.copy` resolves both paths through `get_path`. Local and S3 `get` read bytes and return `str` for UTF-8 content or raw `bytes` otherwise, and a new `get_bytes` (also on `StorageManager`/`Storage`) always returns bytes. `get` returns `bytes` for any non-UTF-8 content, including latin-1 text. `S3Driver.append` keeps the existing content. Vite `use_*_tag_attributes` accept dicts and `False`; the self-referencing lambda is gone. A pure ASGI `ViteMiddleware` (registered by `ViteProvider`) opens `Vite.request_scope()`, which holds the request's CSP nonce and preloaded assets in a per-instance `ContextVar` shared with child tasks. Outside a scope, the instance values are process-wide defaults, so a nonce set at startup still applies, and middleware inside the scope can read preloaded assets after `call_next`. `Inertia.set_root_view`/`Inertia.version` write to the current `InertiaRequestState` inside a request and to global defaults otherwise. | +| [041](041-database-install-command.md) | 2026-10-22 | InstallCommand base class and database:install | `console/install_command.py` adds an `InstallCommand` base class with a fluent class builder: `InstallCommand.register(name).copy(stub, dest).add_provider(Provider, "config.x.XConfig")`. Each call returns a new subclass. Providers and configs can be classes or dotted strings (strings for app-side configs and optional extras). `--force`, `publish_file` (skips existing files unless forced) and `register_provider` are shared. `register_provider` finds `Application(...)` in `bootstrap/application.py` with `ast` and inserts the entry and any missing imports at AST-reported positions, so formatting is kept. It does nothing if the provider is already listed, and prints manual steps and exits 1 if the bootstrap file is missing. `database:install` (registered by `AppProvider`) subclasses it to add `--driver` and render `config/database.py` from the installer stub. Both live in `console/` so the optional SQLAlchemy extra is not imported. | diff --git a/fastapi_startkit/src/fastapi_startkit/console/database_install_command.py b/fastapi_startkit/src/fastapi_startkit/console/database_install_command.py new file mode 100644 index 00000000..5e517dc6 --- /dev/null +++ b/fastapi_startkit/src/fastapi_startkit/console/database_install_command.py @@ -0,0 +1,40 @@ +from importlib.resources import files +from string import Template +from typing import ClassVar + +from cleo.helpers import option +from cleo.io.inputs.option import Option + +from fastapi_startkit.console.install_command import InstallCommand, ProviderEntry +from fastapi_startkit.installer.plan import DRIVER_CLASSES + + +class DatabaseInstallCommand(InstallCommand): + name = "database:install" + description = "Publish config/database.py and register DatabaseProvider in bootstrap/application.py." + + providers = (ProviderEntry.of("fastapi_startkit.masoniteorm.DatabaseProvider", "config.database.DatabaseConfig"),) + + options: ClassVar[list[Option]] = [ + *InstallCommand.options, + option( + "driver", + "d", + description=f"Database driver: {', '.join(DRIVER_CLASSES)}.", + flag=False, + default="sqlite", + ), + ] + + def publish(self) -> int: + driver = self.option("driver") + if driver not in DRIVER_CLASSES: + self.line_error(f"Unsupported driver '{driver}'. Use one of: {', '.join(DRIVER_CLASSES)}.") + return 1 + + stub = files("fastapi_startkit.installer").joinpath("stubs/database/config.py.stub").read_text(encoding="utf-8") + self.publish_file( + "config/database.py", + Template(stub).substitute(driver=driver, driver_class=DRIVER_CLASSES[driver]), + ) + return 0 diff --git a/fastapi_startkit/src/fastapi_startkit/console/install_command.py b/fastapi_startkit/src/fastapi_startkit/console/install_command.py new file mode 100644 index 00000000..aaf6f22c --- /dev/null +++ b/fastapi_startkit/src/fastapi_startkit/console/install_command.py @@ -0,0 +1,200 @@ +import ast +from collections.abc import Mapping, Sequence +from dataclasses import dataclass +from importlib.resources.abc import Traversable +from os import PathLike +from pathlib import Path +from typing import ClassVar, Self, cast + +from cleo.helpers import option +from cleo.io.inputs.option import Option + +from fastapi_startkit.console.command import Command + +BOOTSTRAP_PATH = "bootstrap/application.py" + +Insert = tuple[int, str] +Stub = str | PathLike[str] | Traversable + + +def dotted_path(reference: type | str) -> str: + return reference if isinstance(reference, str) else f"{reference.__module__}.{reference.__qualname__}" + + +@dataclass(frozen=True) +class ProviderEntry: + provider: str + config: str | None = None + + @classmethod + def of(cls, provider: type | str, config: type | str | None = None) -> "ProviderEntry": + return cls(dotted_path(provider), dotted_path(config) if config is not None else None) + + @property + def name(self) -> str: + return self.provider.rpartition(".")[2] + + @property + def entry(self) -> str: + if self.config is None: + return self.name + return f"({self.name}, {self.config.rpartition('.')[2]})" + + @property + def imports(self) -> dict[str, str]: + references = [self.provider] if self.config is None else [self.provider, self.config] + return { + name: f"from {module} import {name}" + for module, _, name in (reference.rpartition(".") for reference in references) + } + + +class InstallCommand(Command): + stubs: ClassVar[tuple[tuple[Stub, str], ...]] = () + providers: ClassVar[tuple[ProviderEntry, ...]] = () + + options: ClassVar[list[Option]] = [option("force", "f", description="Overwrite files that already exist.")] + + @classmethod + def register(cls, name: str, description: str = "") -> type[Self]: + return cls.extend(name=name, description=description) + + @classmethod + def copy(cls, stub: Stub, destination: str) -> type[Self]: + return cls.extend(stubs=(*cls.stubs, (stub, destination))) + + @classmethod + def add_provider(cls, provider: type | str, config: type | str | None = None) -> type[Self]: + return cls.extend(providers=(*cls.providers, ProviderEntry.of(provider, config))) + + @classmethod + def extend(cls, **attributes: object) -> type[Self]: + return cast(type[Self], type(cls.__name__, (cls,), attributes)) + + def handle(self) -> int: + return self.publish() or self.register_providers() + + def publish(self) -> int: + for stub, destination in self.stubs: + source = stub if isinstance(stub, Traversable) else Path(stub) + self.publish_file(destination, source.read_text(encoding="utf-8")) + return 0 + + @property + def base_path(self) -> Path: + return Path(self.container.base_path) + + def publish_file(self, relative_path: str, content: str) -> None: + path = self.base_path / relative_path + if path.exists() and not self.option("force"): + self.line(f"{relative_path} already exists, skipped. Use --force to overwrite it.") + return + + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(content, encoding="utf-8") + self.line(f"Created {relative_path}") + + def register_providers(self) -> int: + return max((self.register_provider(provider) for provider in self.providers), default=0) + + def register_provider(self, provider: ProviderEntry) -> int: + path = self.base_path / BOOTSTRAP_PATH + source = path.read_text(encoding="utf-8") if path.exists() else None + updated = ( + register_provider(source, provider.name, provider.entry, provider.imports) if source is not None else None + ) + + if updated is None: + self.line_error(f"Could not find Application(providers=[...]) in {BOOTSTRAP_PATH}.") + self.line_error(f"Register {provider.name} manually:") + for statement in provider.imports.values(): + self.line_error(f" {statement}") + self.line_error(f" providers=[..., {provider.entry}]") + return 1 + + if updated == source: + self.line(f"{provider.name} is already registered in {BOOTSTRAP_PATH}.") + return 0 + + path.write_text(updated, encoding="utf-8") + self.line(f"Registered {provider.name} in {BOOTSTRAP_PATH}") + return 0 + + +def register_provider(source: str, provider: str, entry: str, imports: Mapping[str, str]) -> str | None: + tree = ast.parse(source) + call = find_application_call(tree) + if call is None: + return None + + lines = source.splitlines(keepends=True) + keyword = next((keyword for keyword in call.keywords if keyword.arg == "providers"), None) + + if keyword is None: + inserts = append_item(source, lines, call, [*call.args, *call.keywords], f"providers=[{entry}]") + elif not isinstance(keyword.value, ast.List): + return None + elif any(isinstance(node, ast.Name) and node.id == provider for node in ast.walk(keyword.value)): + return source + else: + inserts = append_item(source, lines, keyword.value, keyword.value.elts, entry) + + inserts.extend(import_inserts(tree, lines, imports)) + for position, text in sorted(inserts, key=lambda insert: insert[0], reverse=True): + source = source[:position] + text + source[position:] + return source + + +def find_application_call(tree: ast.Module) -> ast.Call | None: + for node in ast.walk(tree): + if not isinstance(node, ast.Call): + continue + func = node.func + if (isinstance(func, ast.Name) and func.id == "Application") or ( + isinstance(func, ast.Attribute) and func.attr == "Application" + ): + return node + return None + + +def append_item( + source: str, lines: list[str], node: ast.expr, items: Sequence[ast.expr | ast.keyword], text: str +) -> list[Insert]: + assert node.end_lineno is not None and node.end_col_offset is not None + close = offset(lines, node.end_lineno, node.end_col_offset - 1) + if not items: + return [(close, text)] + + last = max(items, key=lambda item: (item.end_lineno or 0, item.end_col_offset or 0)) + assert last.end_lineno is not None and last.end_col_offset is not None + end = offset(lines, last.end_lineno, last.end_col_offset) + has_comma = source[end:close].lstrip().startswith(",") + + if last.end_lineno == node.end_lineno: + return [(close, (" " if has_comma else ", ") + text)] + + line = lines[last.end_lineno - 1] + line_end = offset(lines, last.end_lineno, 0) + len(line.rstrip("\r\n")) + indent = line[: len(line) - len(line.lstrip())] + inserts = [(line_end, f"\n{indent}{text},")] + if not has_comma: + inserts.append((end, ",")) + return inserts + + +def import_inserts(tree: ast.Module, lines: list[str], imports: Mapping[str, str]) -> list[Insert]: + existing = [node for node in tree.body if isinstance(node, (ast.Import, ast.ImportFrom))] + imported = {alias.asname or alias.name for node in existing for alias in node.names} + missing = [statement for name, statement in imports.items() if name not in imported] + if not missing: + return [] + if not existing: + return [(0, "\n".join(missing) + "\n\n")] + + last = existing[-1].end_lineno + assert last is not None + return [(offset(lines, last, 0) + len(lines[last - 1].rstrip("\r\n")), "\n" + "\n".join(missing))] + + +def offset(lines: list[str], lineno: int, col: int) -> int: + return sum(len(line) for line in lines[: lineno - 1]) + len(lines[lineno - 1].encode()[:col].decode()) diff --git a/fastapi_startkit/src/fastapi_startkit/foundation/app_provider.py b/fastapi_startkit/src/fastapi_startkit/foundation/app_provider.py index 2f43a87f..69e0ac98 100644 --- a/fastapi_startkit/src/fastapi_startkit/foundation/app_provider.py +++ b/fastapi_startkit/src/fastapi_startkit/foundation/app_provider.py @@ -1,3 +1,4 @@ +from fastapi_startkit.console.database_install_command import DatabaseInstallCommand from fastapi_startkit.console.publish_command import PublishCommand from fastapi_startkit.support import Provider @@ -7,4 +8,4 @@ def register(self) -> None: pass def boot(self) -> None: - self.commands([PublishCommand]) + self.commands([PublishCommand, DatabaseInstallCommand]) diff --git a/fastapi_startkit/src/fastapi_startkit/installer/plan.py b/fastapi_startkit/src/fastapi_startkit/installer/plan.py index 00dcf3ec..711ecd3c 100644 --- a/fastapi_startkit/src/fastapi_startkit/installer/plan.py +++ b/fastapi_startkit/src/fastapi_startkit/installer/plan.py @@ -6,6 +6,8 @@ from pathlib import Path from string import Template +DRIVER_CLASSES = {"sqlite": "SQLiteConfig", "postgres": "PostgresConfig", "mysql": "MySQLConfig"} + @dataclass(frozen=True) class InstallOptions: @@ -117,7 +119,7 @@ def define_web() -> Section: def define_database(driver: str, project_name: str) -> Section: if driver == "none": return Section() - driver_class = {"sqlite": "SQLiteConfig", "postgres": "PostgresConfig", "mysql": "MySQLConfig"}[driver] + driver_class = DRIVER_CLASSES[driver] environment = {"DB_CONNECTION": driver} if driver == "sqlite": environment["DB_DATABASE"] = "databases/database.sqlite" diff --git a/fastapi_startkit/tests/console/test_database_install_command.py b/fastapi_startkit/tests/console/test_database_install_command.py new file mode 100644 index 00000000..3a29d6d5 --- /dev/null +++ b/fastapi_startkit/tests/console/test_database_install_command.py @@ -0,0 +1,129 @@ +import ast +from pathlib import Path +from types import SimpleNamespace + +from cleo.testers.command_tester import CommandTester + +from fastapi_startkit.console.database_install_command import DatabaseInstallCommand +from fastapi_startkit.console.install_command import InstallCommand +from fastapi_startkit.foundation.app_provider import AppProvider + +BOOTSTRAP = """from pathlib import Path + +from fastapi_startkit import Application +from fastapi_startkit.logging import LogProvider +from config.app import AppConfig +from config.logging import LoggingConfig + +app = Application( + base_path=Path(__file__).resolve().parent.parent, + config=AppConfig, + providers=[ + (LogProvider, LoggingConfig) # logging + ], +) +""" + + +def run(base_path: Path, args: str = "") -> CommandTester: + command = DatabaseInstallCommand() + command.set_container(SimpleNamespace(base_path=base_path)) # type: ignore[arg-type] + tester = CommandTester(command) + tester.execute(args) + return tester + + +def make_app(tmp_path: Path) -> Path: + (tmp_path / "bootstrap").mkdir() + (tmp_path / "bootstrap/application.py").write_text(BOOTSTRAP) + return tmp_path + + +def providers_of(source: str) -> list[str]: + for node in ast.walk(ast.parse(source)): + if isinstance(node, ast.keyword) and node.arg == "providers": + assert isinstance(node.value, ast.List) + return [ast.unparse(item) for item in node.value.elts] + raise AssertionError("providers keyword not found") + + +def test_is_an_install_command(): + assert issubclass(DatabaseInstallCommand, InstallCommand) + + +def test_installs_config_and_registers_provider(tmp_path): + app = make_app(tmp_path) + + tester = run(app) + + assert tester.status_code == 0 + config = (app / "config/database.py").read_text() + assert 'env("DB_CONNECTION", "sqlite")' in config + assert "SQLiteConfig()" in config + bootstrap = (app / "bootstrap/application.py").read_text() + assert providers_of(bootstrap) == ["(LogProvider, LoggingConfig)", "(DatabaseProvider, DatabaseConfig)"] + assert ( + "from config.logging import LoggingConfig\n" + "from fastapi_startkit.masoniteorm import DatabaseProvider\n" + "from config.database import DatabaseConfig\n" + ) in bootstrap + compile(bootstrap, "application.py", "exec") + output = tester.io.fetch_output() + assert "Created config/database.py" in output + assert "Registered DatabaseProvider in bootstrap/application.py" in output + + +def test_second_run_changes_nothing(tmp_path): + app = make_app(tmp_path) + run(app) + config = (app / "config/database.py").read_text() + bootstrap = (app / "bootstrap/application.py").read_text() + + tester = run(app, "--driver postgres") + + assert tester.status_code == 0 + assert (app / "config/database.py").read_text() == config + assert (app / "bootstrap/application.py").read_text() == bootstrap + output = tester.io.fetch_output() + assert "config/database.py already exists, skipped" in output + assert "DatabaseProvider is already registered" in output + + +def test_force_overwrites_existing_config(tmp_path): + app = make_app(tmp_path) + (app / "config").mkdir() + (app / "config/database.py").write_text("OLD = True\n") + + tester = run(app, "--driver mysql --force") + + assert tester.status_code == 0 + assert "MySQLConfig()" in (app / "config/database.py").read_text() + + +def test_rejects_unknown_driver(tmp_path): + app = make_app(tmp_path) + bootstrap = (app / "bootstrap/application.py").read_text() + + tester = run(app, "--driver oracle") + + assert tester.status_code == 1 + assert "Unsupported driver 'oracle'" in tester.io.fetch_error() + assert not (app / "config").exists() + assert (app / "bootstrap/application.py").read_text() == bootstrap + + +def test_missing_bootstrap_prints_manual_steps(tmp_path): + tester = run(tmp_path) + + assert tester.status_code == 1 + assert (tmp_path / "config/database.py").exists() + assert "from fastapi_startkit.masoniteorm import DatabaseProvider" in tester.io.fetch_error() + + +def test_app_provider_registers_command(): + application = SimpleNamespace(commands=[]) + application.add_commands = application.commands.extend + + AppProvider(application).boot() # type: ignore[arg-type] + + assert DatabaseInstallCommand in application.commands diff --git a/fastapi_startkit/tests/console/test_install_command.py b/fastapi_startkit/tests/console/test_install_command.py new file mode 100644 index 00000000..0137607d --- /dev/null +++ b/fastapi_startkit/tests/console/test_install_command.py @@ -0,0 +1,241 @@ +import ast +from importlib.resources import files +from pathlib import Path +from types import SimpleNamespace + +from cleo.testers.command_tester import CommandTester + +from fastapi_startkit.console.install_command import InstallCommand, ProviderEntry, register_provider +from fastapi_startkit.logging import LogProvider + +IMPORTS = { + "CacheProvider": "from fastapi_startkit.cache import CacheProvider", + "CacheConfig": "from config.cache import CacheConfig", +} + +BOOTSTRAP = """from fastapi_startkit import Application +from config.logging import LoggingConfig + +app = Application( + providers=[ + (LogProvider, LoggingConfig) # logging + ], +) +""" + + +def cache_install(tmp_path: Path) -> type[InstallCommand]: + stub = tmp_path / "cache.py.stub" + stub.write_text("CACHE = True\n") + return ( + InstallCommand.register("cache:install", "Install the cache.") + .copy(stub, "config/cache.py") + .add_provider("fastapi_startkit.cache.CacheProvider", "config.cache.CacheConfig") + ) + + +def provider_only_install() -> type[InstallCommand]: + return InstallCommand.register("broadcast:install").add_provider("fastapi_startkit.broadcasting.BroadcastProvider") + + +def run(command_class: type[InstallCommand], base_path: Path, args: str = "") -> CommandTester: + command = command_class() + command.set_container(SimpleNamespace(base_path=base_path)) # type: ignore[arg-type] + tester = CommandTester(command) + tester.execute(args) + return tester + + +def make_app(tmp_path: Path) -> Path: + (tmp_path / "bootstrap").mkdir() + (tmp_path / "bootstrap/application.py").write_text(BOOTSTRAP) + return tmp_path + + +def register(source: str) -> str | None: + return register_provider(source, "CacheProvider", "(CacheProvider, CacheConfig)", IMPORTS) + + +def providers_of(source: str) -> list[str]: + for node in ast.walk(ast.parse(source)): + if isinstance(node, ast.keyword) and node.arg == "providers": + assert isinstance(node.value, ast.List) + return [ast.unparse(item) for item in node.value.elts] + raise AssertionError("providers keyword not found") + + +def test_subclass_publishes_and_registers(tmp_path): + app = make_app(tmp_path) + + tester = run(cache_install(tmp_path), app) + + assert tester.status_code == 0 + assert (app / "config/cache.py").read_text() == "CACHE = True\n" + bootstrap = (app / "bootstrap/application.py").read_text() + assert " (LogProvider, LoggingConfig), # logging\n (CacheProvider, CacheConfig),\n" in bootstrap + assert ( + "from config.logging import LoggingConfig\n" + "from fastapi_startkit.cache import CacheProvider\n" + "from config.cache import CacheConfig\n" + ) in bootstrap + output = tester.io.fetch_output() + assert "Created config/cache.py" in output + assert "Registered CacheProvider in bootstrap/application.py" in output + + +def test_second_run_changes_nothing(tmp_path): + app = make_app(tmp_path) + run(cache_install(tmp_path), app) + bootstrap = (app / "bootstrap/application.py").read_text() + + tester = run(cache_install(tmp_path), app) + + assert tester.status_code == 0 + assert (app / "bootstrap/application.py").read_text() == bootstrap + output = tester.io.fetch_output() + assert "config/cache.py already exists, skipped" in output + assert "CacheProvider is already registered" in output + + +def test_fluent_chain_builds_independent_command_classes(): + command = InstallCommand.register("log:install", "Install logging.").add_provider( + LogProvider, "config.logging.LoggingConfig" + ) + + assert issubclass(command, InstallCommand) + assert command.name == "log:install" + assert command.description == "Install logging." + assert InstallCommand.providers == () + assert command.providers == ( + ProviderEntry(f"{LogProvider.__module__}.LogProvider", "config.logging.LoggingConfig"), + ) + + +def test_provider_entry_from_class_derives_entry_and_imports(): + entry = ProviderEntry.of(LogProvider, "config.logging.LoggingConfig") + + assert entry.name == "LogProvider" + assert entry.entry == "(LogProvider, LoggingConfig)" + assert entry.imports == { + "LogProvider": f"from {LogProvider.__module__} import LogProvider", + "LoggingConfig": "from config.logging import LoggingConfig", + } + + +def test_registers_multiple_providers_and_copies_traversable_stub(tmp_path): + app = make_app(tmp_path) + stub = files("fastapi_startkit.installer").joinpath("stubs/database/config.py.stub") + command = ( + InstallCommand.register("multi:install") + .copy(stub, "config/raw.py") + .add_provider("package.AProvider") + .add_provider("package.BProvider", "config.b.BConfig") + ) + + tester = run(command, app) + + assert tester.status_code == 0 + assert (app / "config/raw.py").read_text() == stub.read_text() + assert providers_of((app / "bootstrap/application.py").read_text()) == [ + "(LogProvider, LoggingConfig)", + "AProvider", + "(BProvider, BConfig)", + ] + + +def test_force_overwrites_published_file(tmp_path): + app = make_app(tmp_path) + (app / "config").mkdir() + (app / "config/cache.py").write_text("OLD = True\n") + + run(cache_install(tmp_path), app, "--force") + + assert (app / "config/cache.py").read_text() == "CACHE = True\n" + + +def test_provider_without_config_and_without_publish(tmp_path): + app = make_app(tmp_path) + + tester = run(provider_only_install(), app) + + assert tester.status_code == 0 + assert providers_of((app / "bootstrap/application.py").read_text()) == [ + "(LogProvider, LoggingConfig)", + "BroadcastProvider", + ] + assert not (app / "config").exists() + + +def test_missing_bootstrap_prints_manual_steps(tmp_path): + tester = run(cache_install(tmp_path), tmp_path) + + assert tester.status_code == 1 + assert (tmp_path / "config/cache.py").exists() + error = tester.io.fetch_error() + assert "Could not find Application(providers=[...]) in bootstrap/application.py" in error + assert "from fastapi_startkit.cache import CacheProvider" in error + assert "Register CacheProvider manually:" in error + assert "providers=[..., (CacheProvider, CacheConfig)]" in error + + +def test_single_line_providers_list(): + updated = register( + "from fastapi_startkit import Application\n\napp = Application(providers=[LogProvider, Other])\n" + ) + + assert updated is not None + assert "Application(providers=[LogProvider, Other, (CacheProvider, CacheConfig)])" in updated + + +def test_single_line_providers_list_with_trailing_comma_and_empty_list(): + assert "providers=[A, (CacheProvider, CacheConfig)]" in (register("app = Application(providers=[A,])\n") or "") + assert "providers=[(CacheProvider, CacheConfig)]" in (register("app = Application(providers=[])\n") or "") + + +def test_adds_providers_keyword_when_missing(): + updated = register("from fastapi_startkit import Application\n\napp = Application(\n base_path=None\n)\n") + + assert updated is not None + assert " base_path=None,\n providers=[(CacheProvider, CacheConfig)],\n)" in updated + assert providers_of(updated) == ["(CacheProvider, CacheConfig)"] + + +def test_adds_imports_to_file_without_imports(): + updated = register("app = fastapi_startkit.Application()\n") + + assert updated is not None + assert updated.startswith("from fastapi_startkit.cache import CacheProvider\n") + assert "Application(providers=[(CacheProvider, CacheConfig)])" in updated + + +def test_keeps_existing_imports(): + source = ( + "from fastapi_startkit.cache.providers import CacheProvider\n" + "from config.cache import CacheConfig\n" + "app = Application(providers=[])\n" + ) + + updated = register(source) + + assert updated is not None + assert updated.count("import CacheProvider") == 1 + assert updated.count("import CacheConfig") == 1 + + +def test_already_registered_returns_source_unchanged(): + source = "app = Application(providers=[(CacheProvider, CacheConfig)])\n" + + assert register(source) == source + + +def test_unsupported_bootstrap_shapes_return_none(): + assert register("app = object()\n") is None + assert register("app = Application(providers=PROVIDERS)\n") is None + + +def test_edited_bootstrap_compiles(tmp_path): + app = make_app(tmp_path) + + run(cache_install(tmp_path), app) + + compile((app / "bootstrap/application.py").read_text(), "application.py", "exec")