campaign sending prototype

src/govoplan_core/devserver.py

@@ -0,0 +1,222 @@
+from __future__ import annotations
+
+import argparse
+import importlib.util
+import inspect
+import os
+import sys
+from collections.abc import Iterable, Sequence
+from pathlib import Path
+from typing import Any
+
+from govoplan_core.core.discovery import iter_module_entry_points
+from govoplan_core.core.modules import ModuleManifest
+from govoplan_core.core.registry import PlatformRegistry
+from govoplan_core.server.config import GovoplanServerConfig, import_object, load_server_config
+from govoplan_core.server.registry import build_platform_registry
+
+DEFAULT_APP = "govoplan_core.server.app:app"
+DEFAULT_CONFIG = "govoplan_core.server.default_config:get_server_config"
+
+
+def _config_module_runtime_root(config_path: str | None) -> Path | None:
+    object_path = config_path or os.getenv("GOVOPLAN_SERVER_CONFIG")
+    if not object_path:
+        return None
+    module_name = object_path.split(":", 1)[0]
+    try:
+        spec = importlib.util.find_spec(module_name)
+    except (ImportError, AttributeError, ValueError):
+        return None
+    if spec is None or not spec.origin:
+        return None
+
+    source = Path(spec.origin).resolve()
+    if source.name == "__init__.py":
+        package_dir = source.parent
+    else:
+        package_dir = source.parent
+    if package_dir.name == "app":
+        return package_dir.parent
+    return None
+
+
+def apply_runtime_defaults(config_path: str | None) -> Path | None:
+    runtime_root = _config_module_runtime_root(config_path)
+    if runtime_root is None:
+        return None
+
+    defaults = {
+        "DATABASE_URL": f"sqlite:///{runtime_root / 'multimailer-dev.db'}",
+        "FILE_STORAGE_LOCAL_ROOT": str(runtime_root / "runtime" / "files"),
+        "MOCK_MAILBOX_DIR": str(runtime_root / "runtime" / "mock-mailbox"),
+    }
+    for key, value in defaults.items():
+        os.environ.setdefault(key, value)
+    return runtime_root
+
+
+def _source_root_for_path(path: str | os.PathLike[str]) -> Path | None:
+    source_path = Path(path).resolve()
+    directory = source_path if source_path.is_dir() else source_path.parent
+    if not directory.exists():
+        return None
+
+    for candidate in (directory, *directory.parents):
+        if candidate.name == "src":
+            return candidate
+    return directory
+
+
+def _source_roots_for_object(value: Any) -> tuple[Path, ...]:
+    candidates: list[Any] = [value]
+    if not inspect.isclass(value) and not inspect.ismodule(value):
+        candidates.append(type(value))
+
+    roots: list[Path] = []
+    for candidate in candidates:
+        try:
+            source = inspect.getsourcefile(candidate) or inspect.getfile(candidate)
+        except (TypeError, OSError):
+            continue
+        root = _source_root_for_path(source)
+        if root is not None:
+            roots.append(root)
+    return tuple(roots)
+
+
+def _manifest_source_roots(manifest: ModuleManifest) -> tuple[Path, ...]:
+    roots: list[Path] = []
+
+    if manifest.route_factory is not None:
+        roots.extend(_source_roots_for_object(manifest.route_factory))
+
+    if manifest.migration_spec is not None and manifest.migration_spec.script_location:
+        root = _source_root_for_path(manifest.migration_spec.script_location)
+        if root is not None:
+            roots.append(root)
+
+    for provider in manifest.resource_acl_providers:
+        roots.extend(_source_roots_for_object(provider))
+
+    for provider in manifest.tenant_summary_providers:
+        roots.extend(_source_roots_for_object(provider))
+
+    for providers in manifest.delete_veto_providers.values():
+        for provider in providers:
+            roots.extend(_source_roots_for_object(provider))
+
+    return tuple(roots)
+
+
+def _entry_point_source_roots(enabled_module_ids: set[str]) -> tuple[Path, ...]:
+    roots: list[Path] = []
+    for entry_point in iter_module_entry_points():
+        loaded = entry_point.load()
+        manifest = loaded() if callable(loaded) else loaded
+        if not isinstance(manifest, ModuleManifest) or manifest.id not in enabled_module_ids:
+            continue
+        roots.extend(_source_roots_for_object(loaded))
+    return tuple(roots)
+
+
+def _config_source_roots(config_path: str | None) -> tuple[Path, ...]:
+    object_path = config_path or os.getenv("GOVOPLAN_SERVER_CONFIG") or DEFAULT_CONFIG
+    try:
+        loaded = import_object(object_path)
+    except ModuleNotFoundError:
+        if config_path or os.getenv("GOVOPLAN_SERVER_CONFIG"):
+            raise
+        loaded = import_object(DEFAULT_CONFIG)
+    return _source_roots_for_object(loaded)
+
+
+def unique_existing_dirs(paths: Iterable[Path | str]) -> list[str]:
+    seen: set[Path] = set()
+    result: list[str] = []
+    for item in paths:
+        path = Path(item).resolve()
+        if path in seen or not path.is_dir():
+            continue
+        seen.add(path)
+        result.append(str(path))
+    return result
+
+
+def build_reload_dirs(
+    config: GovoplanServerConfig,
+    *,
+    config_path: str | None = None,
+    registry: PlatformRegistry | None = None,
+    extra_dirs: Sequence[str] = (),
+) -> list[str]:
+    active_registry = registry or build_platform_registry(config.enabled_modules, manifest_factories=config.manifest_factories)
+    manifests = active_registry.manifests()
+    enabled_module_ids = {manifest.id for manifest in manifests}
+
+    roots: list[Path | str] = []
+    roots.extend(_config_source_roots(config_path))
+
+    for factory in config.manifest_factories:
+        roots.extend(_source_roots_for_object(factory))
+
+    roots.extend(_entry_point_source_roots(enabled_module_ids))
+
+    for manifest in manifests:
+        roots.extend(_manifest_source_roots(manifest))
+
+    roots.extend(extra_dirs)
+    return unique_existing_dirs(roots)
+
+
+def parse_args(argv: Sequence[str] | None = None) -> argparse.Namespace:
+    parser = argparse.ArgumentParser(description="Run the GovOPlaN development server with module-aware reload directories.")
+    parser.add_argument("--config", help="GovoplanServerConfig object path, e.g. app.govoplan_config:get_server_config.")
+    parser.add_argument("--app", default=DEFAULT_APP, help=f"ASGI app import string passed to uvicorn. Default: {DEFAULT_APP}.")
+    parser.add_argument("--host", default="127.0.0.1", help="Host to bind. Default: 127.0.0.1.")
+    parser.add_argument("--port", type=int, default=8000, help="Port to bind. Default: 8000.")
+    parser.add_argument("--no-reload", action="store_true", help="Disable uvicorn reload.")
+    parser.add_argument("--reload-dir", action="append", default=[], help="Additional directory to watch. May be passed multiple times.")
+    return parser.parse_args(argv)
+
+
+def main(argv: Sequence[str] | None = None) -> int:
+    args = parse_args(argv)
+    if args.config:
+        os.environ["GOVOPLAN_SERVER_CONFIG"] = args.config
+
+    config_path = args.config or os.getenv("GOVOPLAN_SERVER_CONFIG")
+    runtime_root = apply_runtime_defaults(config_path)
+    config = load_server_config(config_path)
+    registry = build_platform_registry(config.enabled_modules, manifest_factories=config.manifest_factories)
+    reload_dirs = build_reload_dirs(config, config_path=config_path, registry=registry, extra_dirs=args.reload_dir)
+
+    print(f"Starting GovOPlaN dev server: {args.app}")
+    print(f"Config: {config_path or DEFAULT_CONFIG}")
+    if runtime_root is not None:
+        print(f"Runtime root: {runtime_root}")
+    print("Modules: " + ", ".join(manifest.id for manifest in registry.manifests()))
+    if args.no_reload:
+        print("Reload: disabled")
+    else:
+        print("Reload dirs:")
+        for directory in reload_dirs:
+            print(f"  - {directory}")
+
+    try:
+        import uvicorn
+    except ModuleNotFoundError as exc:
+        raise SystemExit("uvicorn is not installed. Install govoplan-core with the server extra or requirements-dev.txt.") from exc
+
+    uvicorn.run(
+        args.app,
+        host=args.host,
+        port=args.port,
+        reload=not args.no_reload,
+        reload_dirs=reload_dirs if not args.no_reload else None,
+    )
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main(sys.argv[1:]))

src/govoplan_core/security/module_permissions.py

@@ -39,6 +39,7 @@ LEGACY_TO_MODULE_SCOPES: dict[str, str] = {
     "mail_servers:read": "mail:profile:read",
     "mail_servers:use": "mail:profile:use",
     "mail_servers:test": "mail:profile:test",
+    "mail_servers:mailbox_read": "mail:mailbox:read",
     "mail_servers:write": "mail:profile:write",
     "mail_servers:manage_credentials": "mail:secret:manage",
 }

src/govoplan_core/settings.py

@@ -19,6 +19,7 @@ class Settings(BaseSettings):
     tenant_schema_template: str | None = Field(default=None, alias="TENANT_SCHEMA_TEMPLATE")
     enabled_modules: str = Field(default="access,campaigns,files,mail", alias="ENABLED_MODULES")
     redis_url: str = Field(default="redis://redis:6379/0", alias="REDIS_URL")
+    celery_enabled: bool = Field(default=False, alias="CELERY_ENABLED")
 
     s3_endpoint_url: str = Field(default="http://garage:3900", alias="S3_ENDPOINT_URL")
     s3_region: str = Field(default="garage", alias="S3_REGION")
@@ -30,6 +31,7 @@ class Settings(BaseSettings):
     # production can switch to Garage/S3 without changing API contracts.
     file_storage_backend: str = Field(default="local", alias="FILE_STORAGE_BACKEND")
     file_storage_local_root: str = Field(default="runtime/files", alias="FILE_STORAGE_LOCAL_ROOT")
+    file_storage_local_fallback_roots: str = Field(default="", alias="FILE_STORAGE_LOCAL_FALLBACK_ROOTS")
     file_storage_s3_endpoint_url: str | None = Field(default=None, alias="FILE_STORAGE_S3_ENDPOINT_URL")
     file_storage_s3_region: str | None = Field(default=None, alias="FILE_STORAGE_S3_REGION")
     file_storage_s3_access_key_id: str | None = Field(default=None, alias="FILE_STORAGE_S3_ACCESS_KEY_ID")