diff --git a/.github/workflows/changelog.yml b/.github/workflows/changelog.yml index dbe2044..45a7fa9 100644 --- a/.github/workflows/changelog.yml +++ b/.github/workflows/changelog.yml @@ -11,12 +11,18 @@ permissions: jobs: changelog: runs-on: ubuntu-latest + steps: - - uses: actions/checkout@v4 + # Checkout the actual branch instead of leaving the repository + # in a detached HEAD state. + - name: Checkout repository + uses: actions/checkout@v4 with: fetch-depth: 0 + ref: ${{ github.head_ref || github.ref_name }} - - uses: actions/setup-python@v5 + - name: Setup Python + uses: actions/setup-python@v5 with: python-version: "3.12" @@ -24,7 +30,8 @@ jobs: env: GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }} GITHUB_REPOSITORY: ${{ github.repository }} - run: python scripts/generate_changelog.py --output CHANGELOG.md + run: | + python scripts/generate_changelog.py --output CHANGELOG.md - name: Commit and push changes run: | @@ -32,8 +39,12 @@ jobs: echo "No changelog changes to commit." exit 0 fi + git config user.name "github-actions[bot]" git config user.email "github-actions[bot]@users.noreply.github.com" + git add CHANGELOG.md + git commit -m "docs: sync CHANGELOG.md from GitHub releases" - git push + + git push origin HEAD:${{ github.ref_name }} diff --git a/example/example.py b/example/example.py index 3a480a4..d2d4eb2 100644 --- a/example/example.py +++ b/example/example.py @@ -716,11 +716,11 @@ async def seed_demo_data(session: AsyncSession) -> None: user1 = User( email="alice@example.com", full_name="Alice Johnson", is_active=True, - hashed_password=password_manager.hash("alice"), + password=password_manager.hash("alice"), ) user2 = User( email="bob@example.com", full_name="Bob Smith", is_active=True, - hashed_password=password_manager.hash("bob"), + password=password_manager.hash("bob"), ) session.add_all([user1, user2]) await session.flush() @@ -755,7 +755,7 @@ async def seed_admin_user(session: AsyncSession) -> None: hashed = bcrypt.hashpw(b"admin", bcrypt.gensalt()).decode() admin_user = User( email="admin@example.com", - hashed_password=hashed, + password=hashed, full_name="Admin", is_superuser=True, is_active=True, @@ -824,13 +824,6 @@ async def lifespan(app: FastAPI): {"label": "Settings", "url": "/admin/users/", "icon": "cog-6-tooth"}, {"label": "Help", "url": "https://docs.example.com"}, ], - # Theme configuration - theme=ThemeConfig( - preset="paper", - primary_color="#6366F1", - show_grain_texture=False, - show_accent_line=True, - ), # UI component configuration sidebar_style="compact", table_style="striped", diff --git a/example/example_ai.py b/example/example_ai.py index 3fa37e1..577600c 100644 --- a/example/example_ai.py +++ b/example/example_ai.py @@ -835,7 +835,7 @@ async def seed_admin(session: AsyncSession) -> None: hashed = bcrypt.hashpw(b"admin", bcrypt.gensalt()).decode() admin_user = User( email="admin@example.com", - hashed_password=hashed, + password=hashed, full_name="Admin", is_superuser=True, is_active=True, diff --git a/example/example_custom_templates.py b/example/example_custom_templates.py index 593092c..00624ba 100644 --- a/example/example_custom_templates.py +++ b/example/example_custom_templates.py @@ -165,7 +165,7 @@ async def lifespan(app: FastAPI): session.add( User( email="admin@example.com", - hashed_password=hashed, + password=hashed, full_name="Admin", is_superuser=True, is_active=True, diff --git a/example/example_sqlmodel.py b/example/example_sqlmodel.py index 009eb7a..2ae5f22 100644 --- a/example/example_sqlmodel.py +++ b/example/example_sqlmodel.py @@ -220,7 +220,7 @@ async def lifespan(app: FastAPI): hashed = bcrypt.hashpw(b"admin", bcrypt.gensalt()).decode() admin_user = User( email="admin@example.com", - hashed_password=hashed, + password=hashed, full_name="Admin", is_superuser=True, is_active=True, diff --git a/fastapi_admin_kit/__init__.py b/fastapi_admin_kit/__init__.py index 645de87..211698c 100644 --- a/fastapi_admin_kit/__init__.py +++ b/fastapi_admin_kit/__init__.py @@ -134,4 +134,4 @@ "configure_notifications", "notifications_router", ] -__version__ = "0.5.0" +__version__ = "0.5.1" diff --git a/fastapi_admin_kit/admin/admin_database.py b/fastapi_admin_kit/admin/admin_database.py index d0d0707..308a5c5 100644 --- a/fastapi_admin_kit/admin/admin_database.py +++ b/fastapi_admin_kit/admin/admin_database.py @@ -94,6 +94,21 @@ async def _create_tables( exclude = set(extra_exclude_tables or ()) + # If the configured backend cloned the project's auth_model table + # into AdminBase.metadata (so its FK can resolve within the same + # MetaData), drop the clone from the create_all table list — the + # project's own metadata owns the real table and DDL for it + # would conflict. Check both the backend and the database + # instance for the cloned names (the backend stores them on the + # database when available, which is the more reliable path for + # tests that swap the backend). + cloned = set() + backend = getattr(self, "_backend", None) + if backend is not None: + cloned = getattr(backend, "_cloned_auth_tables", set()) or set() + cloned |= getattr(self, "_cloned_auth_tables", set()) or set() + exclude |= cloned + def _filtered(metadata: Any) -> Any: drop = exclude | (set() if include_ai_tables else AI_TABLE_NAMES) if not drop: diff --git a/fastapi_admin_kit/admin/core.py b/fastapi_admin_kit/admin/core.py index 05e5200..bd5eb94 100644 --- a/fastapi_admin_kit/admin/core.py +++ b/fastapi_admin_kit/admin/core.py @@ -1162,6 +1162,32 @@ def _attr(obj: Any, name: str) -> Any: return getattr(obj, name, "") self._jinja_env.env.filters["slugify"] = slugify + + def file_url(path: Any) -> str: + """Map a stored file path to its public URL for ``src``/``href``. + + Storage keeps relative paths (``"documents/uuid.jpg"``) so that + ``upload_dir / path`` and ``delete(path)`` work. Browsers need + an absolute public URL (``"/uploads/documents/uuid.jpg"``), so + templates must pipe stored values through this filter instead + of rendering them raw. + """ + if not path: + return "" + s = str(path) + if s.startswith(("http://", "https://", "data:", "blob:")): + return s + storage = getattr(getattr(self.config, "storage", None), "storage", None) + if storage is not None and hasattr(storage, "url"): + try: + return storage.url(s) + except Exception: + pass + storage_cfg = getattr(self.config, "storage", None) + base = (getattr(storage_cfg, "uploads_url", None) or "/uploads").rstrip("/") + return f"{base}/{s.lstrip('/')}" + + self._jinja_env.env.filters["file_url"] = file_url self._jinja_env.env.globals["attr"] = _attr from fastapi_admin_kit.inspection import model_display_name @@ -1481,6 +1507,18 @@ def _builtin_user_tables_to_skip(self) -> tuple[str, ...]: """Return the names of built-in tables that should NOT be created when a custom ``auth_model`` is configured. + When the project supplies its own user model (``auth_model=``), the + built-in ``admin_users`` table is skipped — the custom auth_model + is the source of truth for user identity. ``admin_user_roles`` is + kept and its ``user_id`` foreign key is retargeted to the custom + auth_model's table at create time (see + ``SqlAlchemyDatabaseBackend.adapt_auth_model``) so the junction + links roles to the project's own user rows. + + The role/permission system (``admin_roles``, + ``admin_role_permissions``, ``admin_permissions``) is kept so the + admin can still manage granular per-table access. + Returns an empty tuple when the built-in ``User`` is in use (default installation) or when no ``auth_model`` is configured. """ @@ -1489,7 +1527,7 @@ def _builtin_user_tables_to_skip(self) -> tuple[str, ...]: auth_model = self.config.auth.auth_model if auth_model is None or auth_model is BuiltinUser: return () - return ("admin_users", "admin_user_roles") + return ("admin_users",) async def create_tables(self) -> None: """Create all admin database tables, correctly handling a custom @@ -1522,6 +1560,27 @@ async def create_tables(self) -> None: logger.info("SKIP_CREATE_TABLES=true: skipping admin table creation") return self._adapt_builtin_user_id_columns() + # Ask the configured backend to retarget the built-in user + # relations (admin_user_roles FK, Role.users M2M, etc.) at the + # custom auth_model. Each backend implements this against its own + # ORM primitives; non-SQLA backends may no-op. + # + # Use ``self.database._backend`` (the backend that actually performs + # DDL in ``_create_tables`` below) rather than the composite + # ``self.backend.database`` so tests that swap + # ``admin.database._backend`` with a fake do not mutate the global + # SQLAlchemy mapper state. A backend without ``adapt_auth_model`` + # (e.g. a test fake or memory backend) is treated as no-op. + database_backend = getattr(self.database, "_backend", None) + adapt = getattr(database_backend, "adapt_auth_model", None) + if adapt is not None and self.config.auth.auth_model is not None: + try: + from fastapi_admin_kit.migrations.models import User as BuiltinUser + + if self.config.auth.auth_model is not BuiltinUser: + adapt(self.config.auth.auth_model) + except Exception as exc: # pragma: no cover - defensive + logger.warning("adapt_auth_model failed: %s", exc) extra_exclude = list(self._builtin_user_tables_to_skip()) if extra_exclude: logger.info( @@ -1607,6 +1666,18 @@ def _adapt_builtin_user_id_columns(self) -> None: ) col.type = new_type + # Note: the built-in ``admin_users`` table is excluded from + # ``create_all`` at the call site in ``setup()`` via + # ``AdminDatabase._create_tables(extra_exclude_tables=...)`` — we do + # NOT remove it from ``AdminBase.metadata`` here because the + # ``FacadeDict`` exposed by ``Base.metadata`` is immutable. + # + # FK retargeting on ``admin_user_roles`` and the ``Role.users`` M2M + # re-binding are delegated to the configured backend via + # ``backend.adapt_auth_model(auth_model)`` — this keeps ``core.py`` + # ORM-agnostic. Backends (SQLAlchemy) implement the actual column + # and relationship mutations; memory/no-op backends can ignore. + # Note: the built-in ``admin_users`` and ``admin_user_roles`` tables # are excluded from ``create_all`` at the call site in ``setup()`` via # ``AdminDatabase._create_tables(extra_exclude_tables=...)`` — we do @@ -1625,12 +1696,21 @@ def _excluded_builtin_tables(self) -> frozenset[str]: three user-facing AI tables so they never leak into the sidebar/routes. When notifications are disabled, also excludes the notification tables so the "notifications" sidebar group never appears. + + When a custom ``auth_model`` is configured, also excludes the built-in + ``admin_users`` model from auto-discovery (it is never created when a + custom auth_model is set — see ``_builtin_user_tables_to_skip``). The + ``admin_user_roles`` junction table is NOT excluded: it is kept and its + ``user_id`` foreign key is retargeted to the custom auth_model's table + so role relationships still work end-to-end. """ excluded = set(INTERNAL_TABLE_NAMES) # incl. admin_ai_attachments if not self._ai_enabled: excluded |= AI_TABLE_NAMES if not self._enable_notification: excluded |= NOTIFICATION_TABLE_NAMES + if self._builtin_user_tables_to_skip(): + excluded |= {"admin_users"} return frozenset(excluded) def _add_ai_nav_group(self) -> None: diff --git a/fastapi_admin_kit/auth/backend.py b/fastapi_admin_kit/auth/backend.py index ac7292f..707c77f 100644 --- a/fastapi_admin_kit/auth/backend.py +++ b/fastapi_admin_kit/auth/backend.py @@ -137,8 +137,6 @@ async def authenticate( session = self._resolve_session(session) model = self._get_model() field = getattr(model, login_field, None) - print("model: ", model) - print("field: ", field) if field is None: field = getattr(model, "email", None) if field is None: diff --git a/fastapi_admin_kit/auth/mixins.py b/fastapi_admin_kit/auth/mixins.py index 8836ec7..f44e28e 100644 --- a/fastapi_admin_kit/auth/mixins.py +++ b/fastapi_admin_kit/auth/mixins.py @@ -63,6 +63,24 @@ def role_ids(self) -> list[int]: return [] return [r.id for r in roles] + def __str__(self) -> str: + """Return a human-readable identifier for the user. + + Prefers ``email``, falls back to ``username``, then to ``id``. + Prevents admin templates from dumping raw SQLAlchemy ``__repr__`` + output when the inheriting model does not define ``__str__``. + """ + email = getattr(self, "email", None) + if email: + return str(email) + username = getattr(self, "username", None) + if username: + return str(username) + return str(getattr(self, "id", "")) + + def __repr__(self) -> str: + return f"<{type(self).__name__} {self.__str__()}>" + def verify_password(self, password: str) -> bool: """Check if plaintext password matches the stored hash.""" from fastapi_admin_kit.auth.password import password_manager diff --git a/fastapi_admin_kit/auth/views.py b/fastapi_admin_kit/auth/views.py index 522848e..0fffa39 100644 --- a/fastapi_admin_kit/auth/views.py +++ b/fastapi_admin_kit/auth/views.py @@ -120,9 +120,8 @@ async def login_post( login_field=login_field, query_adapter=query_adapter, ) - print("login user: ", user) + except TypeError: - print("login user error: ", user) # Custom backends that don't accept query_adapter user = await auth_backend.authenticate(username, password, session, login_field=login_field) if user is not None: diff --git a/fastapi_admin_kit/backends/memory.py b/fastapi_admin_kit/backends/memory.py index f8752b8..7028ac1 100644 --- a/fastapi_admin_kit/backends/memory.py +++ b/fastapi_admin_kit/backends/memory.py @@ -565,6 +565,16 @@ def _init(self: Any, **kwargs: Any) -> None: def session_adapter_class(self) -> type: return MemorySessionBackend + def adapt_auth_model(self, auth_model: Any) -> None: + """No-op for the in-memory backend. + + Memory backend schemas are flat dicts; there is no foreign-key or + relationship machinery to retarget. Custom auth_models work as-is + because the auth layer always queries through the model attribute + lookup, not the table. + """ + return None + # --------------------------------------------------------------------------- # Composite backend diff --git a/fastapi_admin_kit/backends/protocols.py b/fastapi_admin_kit/backends/protocols.py index f4ccb3f..573ec45 100644 --- a/fastapi_admin_kit/backends/protocols.py +++ b/fastapi_admin_kit/backends/protocols.py @@ -278,3 +278,21 @@ def session_adapter_class(self) -> type: directly and never needs this. """ ... + + def adapt_auth_model(self, auth_model: type) -> None: + """Retarget built-in user relations at a custom ``auth_model``. + + Called once during ``Admin.create_tables()`` when a custom + ``auth_model`` is configured. Backends should: + + - retarget the built-in ``admin_user_roles`` user-side foreign key to + the auth_model's table, + - mirror the auth_model's primary-key type onto the + ``admin_user_roles.user_id`` column, + - rebind any M2M relationships that pointed at the built-in + ``admin_users`` model (e.g. ``Role.users``) so joins route to the + custom auth_model. + + Non-SQLA backends (memory, future ODMs) may no-op. + """ + ... diff --git a/fastapi_admin_kit/backends/sqlalchemy.py b/fastapi_admin_kit/backends/sqlalchemy.py index 2b94101..f5a0d79 100644 --- a/fastapi_admin_kit/backends/sqlalchemy.py +++ b/fastapi_admin_kit/backends/sqlalchemy.py @@ -11,6 +11,7 @@ from __future__ import annotations +import logging from collections.abc import Sequence from typing import TYPE_CHECKING, Any @@ -21,6 +22,186 @@ if TYPE_CHECKING: from fastapi_admin_kit.admin.admin_database import AdminDatabase +logger = logging.getLogger(__name__) + +# --------------------------------------------------------------------------- +# Custom auth_model adaptation state. +# +# ``SqlAlchemyDatabaseBackend.adapt_auth_model`` mutates *global* SQLAlchemy +# state (the ``admin_user_roles`` FK, ``Role.users`` / ``User.roles`` mapper +# properties, log-pattern ``user_id`` column types, and a shadow table in +# ``AdminBase.metadata``). In production this runs once per process (a single +# ``auth_model``), but the test-suite exercises several different +# ``auth_model`` classes in one pytest session. Without a snapshot/restore +# mechanism the first adaptation permanently rebinds the global ``Role`` +# mapper onto a per-test class, breaking every later test that creates a +# built-in ``User`` (``AttributeError: 'NoneType' object has no attribute +# 'get_all_pending'`` / ``NoForeignKeysError``). +# +# ``_ADAPT_SNAPSHOT`` holds the pristine objects captured on the first +# adaptation so :meth:`SqlAlchemyDatabaseBackend.reset_auth_model_adaptation` +# (and the module-level :func:`reset_builtin_auth_adaptation`) can restore +# them. Objects — not target-name strings — are stored: re-creating +# ``ForeignKey(target)`` from a string yields an orphan FK with no parent +# column, which breaks mapper configuration. +# --------------------------------------------------------------------------- + +_ADAPT_SNAPSHOT: dict[str, Any] | None = None +_ADAPTED_AUTH_TABLE: str | None = None + +_LOG_TABLES_WITH_USER_ID: tuple[str, ...] = ( + "admin_audit_log", + "admin_user_permissions", + "admin_refresh_tokens", + "admin_user_totp", + "admin_notifications", + "admin_notification_preferences", + "admin_notification_logs", + "admin_ai_usage_log", + "admin_ai_conversations", +) + + +def _snapshot_builtin_auth_state() -> dict[str, Any]: + """Capture the pristine global auth state before the first adaptation.""" + from fastapi_admin_kit.migrations.models import Role + from fastapi_admin_kit.migrations.models import User as BuiltinUser + + metadata = BuiltinUser.__table__.metadata + junction = metadata.tables.get("admin_user_roles") + snap: dict[str, Any] = { + "role_users_prop": Role.__mapper__._props.get("users"), + "user_roles_prop": BuiltinUser.__mapper__._props.get("roles"), + "log_col_types": {}, + "junction_constraints": None, + "junction_fks": None, + "col_fks": None, + "col_type": None, + } + if junction is not None and "user_id" in junction.c: + col = junction.c["user_id"] + snap["junction_constraints"] = set(junction.constraints) + snap["junction_fks"] = set(junction.foreign_keys) + snap["col_fks"] = set(col.foreign_keys) + snap["col_type"] = col.type + for table_name in _LOG_TABLES_WITH_USER_ID: + table = metadata.tables.get(table_name) + if table is not None and "user_id" in table.c: + snap["log_col_types"][table_name] = table.c["user_id"].type + return snap + + +def reset_builtin_auth_adaptation() -> None: + """Restore the pristine global auth state captured before adaptation. + + No-op when no adaptation has happened. Safe to call multiple times. + Used by the test-suite to isolate custom-``auth_model`` tests from the + rest of the session. + """ + global _ADAPT_SNAPSHOT, _ADAPTED_AUTH_TABLE + if _ADAPT_SNAPSHOT is None: + return + snap = _ADAPT_SNAPSHOT + try: + from sqlalchemy import ForeignKeyConstraint + + from fastapi_admin_kit.migrations.models import Role + from fastapi_admin_kit.migrations.models import User as BuiltinUser + + metadata = BuiltinUser.__table__.metadata + junction = metadata.tables.get("admin_user_roles") + + # 1. Restore junction constraints / FK sets from saved *objects*. + if junction is not None and snap.get("junction_constraints") is not None: + col = junction.c["user_id"] if "user_id" in junction.c else None + for constraint in list(junction.constraints): + if constraint not in snap["junction_constraints"]: + junction.constraints.remove(constraint) + if isinstance(constraint, ForeignKeyConstraint) and col is not None: + for fk in list(constraint.elements): + junction.foreign_keys.discard(fk) + if fk in col.foreign_keys: + col.foreign_keys.discard(fk) + for constraint in snap["junction_constraints"]: + if constraint not in junction.constraints: + junction.constraints.add(constraint) + if snap.get("junction_fks") is not None: + junction.foreign_keys.clear() + junction.foreign_keys.update(snap["junction_fks"]) + if col is not None and snap.get("col_fks") is not None: + col.foreign_keys.clear() + col.foreign_keys.update(snap["col_fks"]) + if col is not None and snap.get("col_type") is not None: + col.type = snap["col_type"] + + # 2. Remove the shadow table created for the custom auth_model. + if _ADAPTED_AUTH_TABLE is not None and _ADAPTED_AUTH_TABLE in metadata.tables: + # Only remove it if it is the minimal shadow (single PK col). + # If the project put its real User table on AdminBase, the table + # predates adaptation and must be kept — but that case never + # creates a shadow (``auth_table_name in metadata.tables``). + # We track creation via the snapshot flag below. + if snap.get("shadow_created") == _ADAPTED_AUTH_TABLE: + try: + metadata.remove(metadata.tables[_ADAPTED_AUTH_TABLE]) + except Exception: + pass + + # 3. Restore log-pattern user_id column types. + for table_name, col_type in snap.get("log_col_types", {}).items(): + table = metadata.tables.get(table_name) + if table is not None and "user_id" in table.c: + table.c["user_id"].type = col_type + + # 4. Restore mapper properties by recreating *fresh* relationships. + # + # Re-adding the saved ``MapperProperty`` instances via + # ``add_property`` does NOT restore ``state.manager[key].impl`` + # (it stays ``None``, surfacing later as ``AttributeError: + # 'NoneType' object has no attribute 'get_all_pending'`` on + # ``session.add(Role(...))``). Fresh ``relationship()`` objects + # instrument correctly. User.roles FIRST: Role.users declares + # ``back_populates="roles"`` and configures immediately, so the + # reverse side must exist or configuration raises + # ``Mapper ... has no property 'roles'``. + if snap.get("user_roles_prop") is not None or snap.get("role_users_prop") is not None: + try: + from sqlalchemy.orm import relationship as _relationship + + junction_table = metadata.tables.get("admin_user_roles") + if junction_table is not None: + try: + BuiltinUser.__mapper__.add_property( + "roles", + _relationship( + Role, + secondary=junction_table, + back_populates="users", + ), + ) + except Exception: + pass + try: + Role.__mapper__.add_property( + "users", + _relationship( + BuiltinUser, + secondary=junction_table, + back_populates="roles", + ), + ) + except Exception: + pass + except Exception: + # Last-resort fallback: restore saved objects directly. + if snap.get("user_roles_prop") is not None: + BuiltinUser.__mapper__._props["roles"] = snap["user_roles_prop"] + if snap.get("role_users_prop") is not None: + Role.__mapper__._props["users"] = snap["role_users_prop"] + finally: + _ADAPT_SNAPSHOT = None + _ADAPTED_AUTH_TABLE = None + def _is_async_session(session: Any) -> bool: """Return True if *session* is an SQLAlchemy async session.""" @@ -775,70 +956,82 @@ def seed_roles( if is_async: async def _run_async() -> None: - existing = await session.all(sa_select(Role)) - if existing and not overwrite: - return - if overwrite: - await session.execute(sa_delete(admin_role_permissions)) - await session.execute(sa_delete(Role)) - for role_spec in seed_roles: - role = Role(name=role_spec.name, description=role_spec.description) - session.add(role) - await session.flush() - await session.refresh(role, ["permissions"]) - if role_spec.permissions: - for table_name, perms in role_spec.permissions.items(): - existing_perm = await session.scalar_one_or_none( - sa_select(Permission).filter_by(table_name=table_name) - ) - if existing_perm is None: - perm = Permission( - name=table_name, - table_name=table_name, - can_view=perms.get("view", False), - can_create=perms.get("create", False), - can_edit=perms.get("edit", False), - can_delete=perms.get("delete", False), + try: + existing = await session.all(sa_select(Role)) + if existing and not overwrite: + return + if overwrite: + await session.execute(sa_delete(admin_role_permissions)) + await session.execute(sa_delete(Role)) + for role_spec in seed_roles: + role = Role(name=role_spec.name, description=role_spec.description) + session.add(role) + await session.flush() + await session.refresh(role, ["permissions"]) + if role_spec.permissions: + for table_name, perms in role_spec.permissions.items(): + existing_perm = await session.scalar_one_or_none( + sa_select(Permission).filter_by(table_name=table_name) ) - session.add(perm) - await session.flush() - else: - perm = existing_perm - role.permissions.append(perm) - await session.commit() + if existing_perm is None: + perm = Permission( + name=table_name, + table_name=table_name, + can_view=perms.get("view", False), + can_create=perms.get("create", False), + can_edit=perms.get("edit", False), + can_delete=perms.get("delete", False), + ) + session.add(perm) + await session.flush() + else: + perm = existing_perm + role.permissions.append(perm) + await session.commit() + finally: + # Always release the session: an unclosed session that + # performed DB work is GC'd while still holding a pooled + # connection ("non-checked-in connection" SAWarning). The + # early-return path above is the usual trigger — it leaves + # an open read transaction behind. close() rolls back any + # open transaction and returns the connection to the pool. + await session.close() return _run_async() - existing = session.all(sa_select(Role)) - if existing and not overwrite: - return None - if overwrite: - session.execute(sa_delete(admin_role_permissions)) - session.execute(sa_delete(Role)) - for role_spec in seed_roles: - role = Role(name=role_spec.name, description=role_spec.description) - session.add(role) - session.flush() - if role_spec.permissions: - for table_name, perms in role_spec.permissions.items(): - existing_perm = session.scalar_one_or_none( - sa_select(Permission).filter_by(table_name=table_name) - ) - if existing_perm is None: - perm = Permission( - name=table_name, - table_name=table_name, - can_view=perms.get("view", False), - can_create=perms.get("create", False), - can_edit=perms.get("edit", False), - can_delete=perms.get("delete", False), + try: + existing = session.all(sa_select(Role)) + if existing and not overwrite: + return None + if overwrite: + session.execute(sa_delete(admin_role_permissions)) + session.execute(sa_delete(Role)) + for role_spec in seed_roles: + role = Role(name=role_spec.name, description=role_spec.description) + session.add(role) + session.flush() + if role_spec.permissions: + for table_name, perms in role_spec.permissions.items(): + existing_perm = session.scalar_one_or_none( + sa_select(Permission).filter_by(table_name=table_name) ) - session.add(perm) - session.flush() - else: - perm = existing_perm - role.permissions.append(perm) - session.commit() + if existing_perm is None: + perm = Permission( + name=table_name, + table_name=table_name, + can_view=perms.get("view", False), + can_create=perms.get("create", False), + can_edit=perms.get("edit", False), + can_delete=perms.get("delete", False), + ) + session.add(perm) + session.flush() + else: + perm = existing_perm + role.permissions.append(perm) + session.commit() + finally: + session.close() return None @property @@ -846,6 +1039,241 @@ def session_adapter_class(self) -> type: """Class wrapping a raw connection into a :class:`SessionBackend`.""" return SqlAlchemySessionAdapter + def adapt_auth_model(self, auth_model: type) -> None: + """Adapt built-in admin metadata for a custom ``auth_model``. + + Called by ``Admin.create_tables()`` when a project supplies its own + user model. Performs the following ORM-specific adaptations so + the built-in ``admin_user_roles`` junction links roles to the + project's own user table (instead of the built-in ``admin_users`` + which is being skipped): + + 1. Creates a minimal "shadow" table reference for the auth_model + inside ``AdminBase.metadata`` (same table name + PK column), + so the FK on ``admin_user_roles.user_id`` can resolve within + the same ``MetaData`` during ``create_all`` sort. The shadow + is NOT created in the database — the project's own metadata + owns the real table. The shadow's name is recorded on the + backend instance so ``AdminDatabase._create_tables`` can drop + it from the ``create_all`` table list. + 2. Retargets the ``admin_user_roles.user_id`` foreign key from + ``admin_users.id`` to ``.`` and + mirrors the PK column type. Also drops the old + ``ForeignKeyConstraint`` from the junction table's + ``constraints`` collection (DDL is emitted from constraints, + not just ``col.foreign_keys``). + 3. Rebinds the ``Role.users`` M2M relationship onto + *auth_model* so ORM joins route to the project user table. + 4. Re-types the built-in log-pattern ``user_id`` columns on + log-style tables (``admin_audit_log``, notifications, etc.) to + match the custom auth_model's primary-key type. + """ + from sqlalchemy import Column as SA_Column + from sqlalchemy import ForeignKeyConstraint + from sqlalchemy import Table as SA_Table + from sqlalchemy import inspect as sa_inspect + from sqlalchemy.orm import relationship + + from fastapi_admin_kit.migrations.models import ( + Role, + ) + from fastapi_admin_kit.migrations.models import ( + User as BuiltinUser, + ) + + if auth_model is BuiltinUser: + return + + global _ADAPT_SNAPSHOT, _ADAPTED_AUTH_TABLE + + metadata = BuiltinUser.__table__.metadata + + pk_cols = sa_inspect(auth_model).primary_key + if not pk_cols: + return + pk_col = pk_cols[0] + new_type = pk_col.type + auth_table_name = auth_model.__tablename__ + + # Idempotency: adapting twice to the same table must not stack + # ``add_property("users", ...)`` calls (each one replaces the + # previous relationship and emits a deprecation warning while + # leaving the class manager in an increasingly inconsistent + # state). If a *different* auth_model was adapted before, reset + # to pristine state first so the new adaptation starts clean. + # + # NOTE: do NOT inspect ``Role.__mapper__._props["users"].mapper`` + # here to decide: accessing ``.mapper`` triggers global mapper + # configuration while the junction is in its retargeted (custom) + # state, which poisons the later reset (User.roles impl stays + # ``None``). The table-name check alone is sufficient. + if _ADAPTED_AUTH_TABLE is not None: + if _ADAPTED_AUTH_TABLE == auth_table_name: + return + reset_builtin_auth_adaptation() + + # Snapshot pristine global state once, before the first mutation. + if _ADAPT_SNAPSHOT is None: + _ADAPT_SNAPSHOT = _snapshot_builtin_auth_state() + _ADAPT_SNAPSHOT["shadow_created"] = None + + # 1. Ensure the auth_model's table is resolvable within + # AdminBase.metadata (needed for sort_tables_and_constraints + # to order admin_user_roles after its FK target). If the table + # is already in admin metadata (e.g. the project puts its User + # model on AdminBase), reuse it. Otherwise create a minimal + # shadow table with the same name + PK column. + if auth_table_name not in metadata.tables: + SA_Table( + auth_table_name, + metadata, + SA_Column(pk_col.name, new_type, primary_key=True), + ) + # Remember that *we* created this table so reset can remove it. + # (If it already existed, it belongs to the project and must be kept.) + if _ADAPT_SNAPSHOT is not None: + _ADAPT_SNAPSHOT["shadow_created"] = auth_table_name + # Record so AdminDatabase._create_tables can drop the shadow from + # the create_all table list (it must not be emitted as DDL). + # Store on the AdminDatabase instance if available so the + # filtering works regardless of backend instance. + db_inst = getattr(self, "_admin_database", None) + if db_inst is not None: + existing = getattr(db_inst, "_cloned_auth_tables", None) + if existing is None: + existing = set() + db_inst._cloned_auth_tables = existing + existing.add(auth_table_name) + if not hasattr(self, "_cloned_auth_tables"): + self._cloned_auth_tables = set() + self._cloned_auth_tables.add(auth_table_name) + + # 2. Retarget the FK on admin_user_roles.user_id + junction = metadata.tables.get("admin_user_roles") + if junction is not None and "user_id" in junction.c: + col = junction.c["user_id"] + + # Drop the old auto-generated ForeignKeyConstraint from the + # table's constraints collection (DDL is emitted from + # constraints, not from col.foreign_keys). Also remove the + # FK objects from the column's foreign_keys set AND from the + # table's foreign_keys set — ``Table.foreign_keys`` is a + # plain ``set`` that aggregates from columns + constraints, + # and stale entries there confuse the M2M relationship + # configuration in step 3. + old_constraints = [ + c + for c in list(junction.constraints) + if isinstance(c, ForeignKeyConstraint) + and any(fk.target_fullname.startswith("admin_users") for fk in c.elements) + ] + for constraint in old_constraints: + junction.constraints.remove(constraint) + for fk in list(constraint.elements): + junction.foreign_keys.discard(fk) + if fk in col.foreign_keys: + col.foreign_keys.discard(fk) + # Belt-and-braces: clear any stale admin_users FK from the + # table-level set. + for fk in list(junction.foreign_keys): + if fk.target_fullname.startswith("admin_users"): + junction.foreign_keys.discard(fk) + + # Add a single ForeignKeyConstraint — its element FK will be + # auto-registered on the column's foreign_keys set, so we do + # NOT add it manually (that would produce duplicates). + new_constraint = ForeignKeyConstraint( + ["user_id"], + [f"{auth_table_name}.{pk_col.name}"], + ondelete="CASCADE", + ) + new_constraint.parent = junction + junction.append_constraint(new_constraint) + col.type = new_type + logger.debug( + "adapt_auth_model: admin_user_roles.user_id -> %s.%s (%r)", + auth_table_name, + pk_col.name, + new_type, + ) + + # 3. Rebind Role.users M2M to the custom auth_model + if hasattr(Role, "users"): + junction_table = metadata.tables.get("admin_user_roles") + auth_has_roles = hasattr(auth_model, "roles") + # Provide explicit primaryjoin/secondaryjoin so SQLAlchemy + # does not have to auto-detect the join across the + # secondary table — the FK on ``admin_user_roles.user_id`` + # now points to a shadow table in admin metadata, which + # makes the auto-detection unreliable. + Role.__mapper__.add_property( + "users", + relationship( + auth_model, + secondary=junction_table, + primaryjoin=Role.id == junction_table.c.role_id, + secondaryjoin=auth_model.__table__.c[pk_col.name] == junction_table.c.user_id, + back_populates="roles" if auth_has_roles else None, + ), + ) + if auth_has_roles: + try: + auth_model.__mapper__.add_property( + "roles", + relationship( + Role, + secondary=junction_table, + primaryjoin=auth_model.__table__.c[pk_col.name] + == junction_table.c.user_id, + secondaryjoin=Role.id == junction_table.c.role_id, + back_populates="users", + ), + ) + except Exception as exc: # pragma: no cover + logger.debug( + "adapt_auth_model: could not add auth_model.roles back-pop: %s", + exc, + ) + logger.debug("adapt_auth_model: Role.users rebound to %s", auth_model) + + # 3b. The built-in User model still has a ``roles`` M2M + # referencing ``admin_user_roles``. Since ``admin_users`` is + # being skipped and the junction's FK now points to the custom + # auth_model, that relationship can no longer auto-resolve. + # Drop it from the mapper so SQLAlchemy does not blow up at + # mapper configuration time. The built-in User is not used in + # the DB when a custom auth_model is configured. + if "roles" in BuiltinUser.__mapper__._props: + BuiltinUser.__mapper__._props.pop("roles", None) + + # 4. Re-type log-pattern user_id columns + for table_name in _LOG_TABLES_WITH_USER_ID: + table = metadata.tables.get(table_name) + if table is None or "user_id" not in table.c: + continue + table.c["user_id"].type = new_type + + _ADAPTED_AUTH_TABLE = auth_table_name + + def reset_auth_model_adaptation(self) -> None: + """Restore pristine global auth state (see :func:`reset_builtin_auth_adaptation`). + + Also clears this instance's (and its ``AdminDatabase``'s, when + available) ``_cloned_auth_tables`` registry so later + ``create_all`` calls stop excluding the stale shadow table. + """ + reset_builtin_auth_adaptation() + try: + self._cloned_auth_tables = set() + except Exception: + pass + try: + db_inst = getattr(self, "_admin_database", None) + if db_inst is not None and hasattr(db_inst, "_cloned_auth_tables"): + db_inst._cloned_auth_tables = set() + except Exception: + pass + def materialize( self, schema: Any, @@ -989,7 +1417,9 @@ def _resolve_target_pk_type(target: Any) -> Any | None: table = md[target] if table is None: if schemas is None: - from fastapi_admin_kit.schemas.builtin import BUILTIN_SCHEMAS + from fastapi_admin_kit.schemas.builtin import ( + BUILTIN_SCHEMAS, + ) reg = schemas if schemas is not None else BUILTIN_SCHEMAS if reg and target in reg: pk = reg[target].get_pk_field() @@ -1096,7 +1526,12 @@ def _resolve_target_pk_type(target: Any) -> Any | None: # Use string-based FK to allow target table to not exist yet columns.append( - Column(f.name, sa_type, ForeignKey(f"{fk_target}.id", use_alter=True), **kwargs) + Column( + f.name, + sa_type, + ForeignKey(f"{fk_target}.id", use_alter=True), + **kwargs, + ) ) else: columns.append(Column(f.name, sa_type, **kwargs)) @@ -1257,7 +1692,12 @@ async def has_perm(self, perm_name: str, session) -> bool: table_name, action = parts attr = f"can_{action}" - if attr not in ("can_view", "can_create", "can_edit", "can_delete"): + if attr not in ( + "can_view", + "can_create", + "can_edit", + "can_delete", + ): return False role_ids = self.role_ids @@ -1279,7 +1719,10 @@ async def has_perm(self, perm_name: str, session) -> bool: result = await session.execute( select(Permission) - .join(UserPermission, UserPermission.permission_id == Permission.id) + .join( + UserPermission, + UserPermission.permission_id == Permission.id, + ) .where(UserPermission.user_id == self.id) ) for perm in result.scalars(): diff --git a/fastapi_admin_kit/cli/user.py b/fastapi_admin_kit/cli/user.py index 6aa49fb..62d372f 100644 --- a/fastapi_admin_kit/cli/user.py +++ b/fastapi_admin_kit/cli/user.py @@ -229,7 +229,7 @@ def _flag(val_col: str | None) -> str: for user in users: name_val = getattr(user, name_col, "") if name_col else "" print( - f"{user.id:<6} {user.email:<30} {str(name_val):<{name_width}} " + f"{str(user.id)[:6]:<6} {user.email:<30} {str(name_val):<{name_width}} " f"{_flag(superuser_col):<10} {_flag(active_col):<8}" ) diff --git a/fastapi_admin_kit/form/pipeline.py b/fastapi_admin_kit/form/pipeline.py index 87e8f0b..770316e 100644 --- a/fastapi_admin_kit/form/pipeline.py +++ b/fastapi_admin_kit/form/pipeline.py @@ -11,6 +11,7 @@ InlineFormsetData, PermissionSet, ) +from fastapi_admin_kit.views.file_handler import FILE_WIDGET_TYPES async def build_inline_formsets( @@ -422,6 +423,10 @@ def build_form_context( fieldsets[0].fields = rendered + has_file_field = any( + isinstance(registered.get_widget(fm.name), FILE_WIDGET_TYPES) + for fm in registered.form_fields + ) return FormContext( model_name=registered.table_name, verbose_name=registered.verbose_name, @@ -436,4 +441,5 @@ def build_form_context( permissions=PermissionSet(), readonly=False, inline_formsets=inline_formsets or [], + has_file_field=has_file_field, ) diff --git a/fastapi_admin_kit/form/types.py b/fastapi_admin_kit/form/types.py index 976d70d..81cd27d 100644 --- a/fastapi_admin_kit/form/types.py +++ b/fastapi_admin_kit/form/types.py @@ -117,6 +117,7 @@ class FormContext: permissions: PermissionSet = field(default_factory=PermissionSet) readonly: bool = False inline_formsets: list[InlineFormsetData] = field(default_factory=list) + has_file_field: bool = False @dataclass diff --git a/fastapi_admin_kit/notifications/dispatcher.py b/fastapi_admin_kit/notifications/dispatcher.py index a301bb7..d2f10f8 100644 --- a/fastapi_admin_kit/notifications/dispatcher.py +++ b/fastapi_admin_kit/notifications/dispatcher.py @@ -18,14 +18,49 @@ import inspect from typing import Any -from sqlalchemy import String, cast, select - from fastapi_admin_kit.db import get_db_session -from fastapi_admin_kit.migrations.models import NotificationPreference, User +from fastapi_admin_kit.migrations.models import NotificationPreference from fastapi_admin_kit.notifications.config import ChangeNotificationConfig from fastapi_admin_kit.notifications.service import NotificationService +def _resolve_user_model(request: Any) -> Any: + """Return the user model that owns ``is_superuser``/``is_active``. + + Prefers the project's ``auth_model`` when one is configured (so joins and + recipient lookups route to the project's user table). Falls back to the + built-in admin ``User`` when no custom auth_model is set. + """ + builtin_user: Any | None + try: + from fastapi_admin_kit.migrations.models import User as BuiltinUser + except Exception: + builtin_user = None + else: + builtin_user = BuiltinUser + + admin = getattr(request.app.state, "admin", None) + auth_model = getattr(admin, "auth_model", None) if admin is not None else None + if auth_model is not None and auth_model is not builtin_user: + return auth_model + return builtin_user + + +def _get_query_backend(request: Any) -> Any: + """Return the configured :class:`QueryBackend` from app state. + + Falls back to importing the SQLAlchemy adapter when the app has not yet + wired the backend (e.g. very early startup paths or test harnesses that + bypass ``Admin()``). + """ + qb = getattr(request.app.state, "admin_query_adapter", None) + if qb is not None: + return qb + from fastapi_admin_kit.backends import SqlAlchemyQueryAdapter + + return SqlAlchemyQueryAdapter() + + async def dispatch_model_change( request: Any, *, @@ -81,14 +116,18 @@ async def dispatch_model_change( # - regular admins only if they have enabled NotificationPreference rows if recipients is None: session = get_db_session(request) + user_model = _resolve_user_model(request) + if user_model is None: + return + qb = _get_query_backend(request) recipients = [] - superusers = await session.all( - select(User).where( - User.is_superuser.is_(True), - User.is_active.is_(True), - ) + superusers_q = qb.where( + qb.select(user_model), + user_model.is_superuser.is_(True), + user_model.is_active.is_(True), ) + superusers = await session.all(superusers_q) for user in superusers: recipients.append( { @@ -99,30 +138,33 @@ async def dispatch_model_change( } ) - pref_user_ids = set( - await session.all( - select(NotificationPreference.user_id).where( - NotificationPreference.enabled.is_(True) - ) - ) + pref_q = qb.where( + qb.select(NotificationPreference), + NotificationPreference.enabled.is_(True), ) + pref_user_ids = { + str(getattr(row, "user_id", None)) + for row in await session.all(pref_q) + if getattr(row, "user_id", None) is not None + } if pref_user_ids: - regular = await session.all( - select(User).where( - User.is_superuser.is_(False), - User.is_active.is_(True), - cast(User.id, String).in_(pref_user_ids), - ) + # Pull active non-superusers and filter by pref in Python so the + # query stays backend-agnostic (no cast()/String literal needed). + regular_q = qb.where( + qb.select(user_model), + user_model.is_superuser.is_(False), + user_model.is_active.is_(True), ) - for user in regular: - recipients.append( - { - "id": getattr(user, "id", None), - "email": getattr(user, "email", None), - "phone": getattr(user, "phone", None), - "channels": cfg.default_channels, - } - ) + for user in await session.all(regular_q): + if str(getattr(user, "id", "")) in pref_user_ids: + recipients.append( + { + "id": getattr(user, "id", None), + "email": getattr(user, "email", None), + "phone": getattr(user, "phone", None), + "channels": cfg.default_channels, + } + ) # Never notify the actor about their own change. if cfg.exclude_actor and actor_id is not None: diff --git a/fastapi_admin_kit/router.py b/fastapi_admin_kit/router.py index e86024a..98db39c 100644 --- a/fastapi_admin_kit/router.py +++ b/fastapi_admin_kit/router.py @@ -616,6 +616,7 @@ async def inline_edit_form( "admin_path": request.app.state.admin_config["admin_path"], "display_columns": form_ctx.fieldsets[0].fields, "inline_fields": form_ctx.fieldsets[0].fields, + "has_file_field": form_ctx.has_file_field, "view": edit_v, }, ) @@ -709,6 +710,7 @@ async def inline_edit_save( "display_columns": form_ctx.fieldsets[0].fields, "inline_fields": form_ctx.fieldsets[0].fields, "errors": errors, + "has_file_field": form_ctx.has_file_field, "view": edit_v, }, status_code=422, diff --git a/fastapi_admin_kit/storage/local.py b/fastapi_admin_kit/storage/local.py index ae17bf3..354aca6 100644 --- a/fastapi_admin_kit/storage/local.py +++ b/fastapi_admin_kit/storage/local.py @@ -95,6 +95,9 @@ async def save(self, file: UploadFile, directory: str = "") -> str: f.write(content) # Return path relative to upload_dir, using forward slashes + # e.g., "filename.jpg" or "directory/filename.jpg". + # Callers build public URLs via ``url()`` and filesystem targets + # via ``upload_dir / path`` — both expect a relative path. if directory: return f"{directory}/{filename}" return filename @@ -108,9 +111,41 @@ async def delete(self, path: str) -> None: if target.is_file(): os.remove(target) - def url(self, path: str) -> str: - """Return the public URL for a stored file.""" - return f"{self.base_url}/{path}" + def url(self, path: str, strip_prefix: bool = False) -> str: + """Return the public URL for a stored file. + + Parameters + ---------- + path : str + The relative path stored in the database. + strip_prefix : bool, default False + When True, strips the leading ``/`` from the URL. + Use ``strip_prefix=True`` when you want the path without + leading slash for ``src`` attributes in templates. + """ + # Accept legacy values stored with a leading "/" or with the + # upload_dir prefix (e.g. "/documents/f.txt" or + # "/tmp/.../uploads/documents/f.txt" from the regressed save()). + # Normal stored values are relative ("documents/f.txt"). + upload_dir_str = str(self.upload_dir) + if upload_dir_str and path.startswith(upload_dir_str + "/"): + path = path[len(upload_dir_str) + 1 :] + elif upload_dir_str and path == upload_dir_str: + path = "" + else: + # Strip a single cosmetic leading slash ("/documents/f.txt" + # -> "documents/f.txt"); "//" is left for the jail check + # to reject downstream. + if path.startswith("/") and not path.startswith("//"): + path = path[1:] + + url = f"{self.base_url}/{path}" + + # Optionally strip leading / from URL + if strip_prefix: + url = url.lstrip("/") + + return url def ensure_dir(self) -> None: """Create the upload directory if it doesn't exist.""" diff --git a/fastapi_admin_kit/templates/admin/base_detail.html b/fastapi_admin_kit/templates/admin/base_detail.html index 458ec8b..3f4cc8b 100644 --- a/fastapi_admin_kit/templates/admin/base_detail.html +++ b/fastapi_admin_kit/templates/admin/base_detail.html @@ -17,7 +17,7 @@ / {{ registered.verbose_name_plural }} / -{{ obj }} +{{ obj.email | default(obj.username, true) | default(obj.id, true) }} {% endblock %} {% block content %} @@ -32,7 +32,8 @@

View {{ registered.verbose_name }}

#{{ obj.id }} - {% if obj.__str__() %} — {{ obj.__str__() }}{% endif %} + {% set _obj_label = obj.email | default(obj.username, true) %} + {% if _obj_label %} — {{ _obj_label }}{% endif %}

diff --git a/fastapi_admin_kit/templates/admin/base_form.html b/fastapi_admin_kit/templates/admin/base_form.html index 5a5b3f8..440870c 100644 --- a/fastapi_admin_kit/templates/admin/base_form.html +++ b/fastapi_admin_kit/templates/admin/base_form.html @@ -22,7 +22,7 @@ / {{ registered.verbose_name_plural }} / -{% if obj %}Edit {{ obj }}{% else %}Create{% endif %} +{% if obj %}{% set _obj_str = obj.email | default(obj.username, true) | default(obj.id, true) %}Edit {{ _obj_str }}{% else %}Create{% endif %} {% endblock %} {% block content %} @@ -40,7 +40,8 @@

{% if obj %}

#{{ obj.id }} - {% if obj.__str__() %} — {{ obj.__str__() }}{% endif %} + {% set _obj_label = obj.email | default(obj.username, true) %} + {% if _obj_label %} — {{ _obj_label }}{% endif %}

{% endif %}

@@ -70,6 +71,7 @@

{% set _ui = ui_config | default({}) %}
diff --git a/fastapi_admin_kit/templates/macros/form_fields.html b/fastapi_admin_kit/templates/macros/form_fields.html index 710dd29..7b3e70d 100644 --- a/fastapi_admin_kit/templates/macros/form_fields.html +++ b/fastapi_admin_kit/templates/macros/form_fields.html @@ -499,13 +499,13 @@ {% set val = field_ctx.widget_context.value | default('') %} {{ _field_wrapper_open(field_ctx) }} {{ _label(field_ctx) }} -
+
{% if val %}
- +
{% endif %}
@@ -533,13 +533,13 @@ {% set val = field_ctx.widget_context.value | default('') %} {{ _field_wrapper_open(field_ctx) }} {{ _label(field_ctx) }} -
+
{% if val %}
- Current: {{ val }} + Current: {{ val }}
{% endif %}
diff --git a/fastapi_admin_kit/templates/partials/detail_field.html b/fastapi_admin_kit/templates/partials/detail_field.html index da82f3e..9b58531 100644 --- a/fastapi_admin_kit/templates/partials/detail_field.html +++ b/fastapi_admin_kit/templates/partials/detail_field.html @@ -30,9 +30,9 @@ {% endfor %} {% if not found and val %}{{ val }}{% endif %} {% elif field.widget_macro == 'image_upload' and val %} - + {% elif field.widget_macro == 'file_upload' and val %} - {{ val }} + {{ val }} {% elif field.widget_macro == 'json_editor' %} 
{{ val }}
{% elif field.widget_macro == 'color_picker' %} diff --git a/fastapi_admin_kit/templates/partials/inline_edit_form.html b/fastapi_admin_kit/templates/partials/inline_edit_form.html index 58a6c58..87d1966 100644 --- a/fastapi_admin_kit/templates/partials/inline_edit_form.html +++ b/fastapi_admin_kit/templates/partials/inline_edit_form.html @@ -4,6 +4,7 @@
str: + """Return a storage-safe directory name derived from *name*. + + Form field names (used as storage subdirectories) must match the + strict ``[A-Za-z0-9_-]`` character set enforced by + :class:`LocalStorageBackend`. This helper maps anything else to a + safe equivalent so field names with spaces, dots, or non-ASCII + characters still work without weakening the security checks at the + storage layer. + """ + sanitized = re.sub(r"[^A-Za-z0-9_-]", "_", name).strip("_") + if not sanitized or not _SAFE_DIRECTORY_RE.match(sanitized): + return "files" + return sanitized + + +def _normalize_stored_path(path: str) -> str: + """Normalize a stored path before passing it back to the storage layer. + + Stored values may have a leading forward-slash (because the + application saves them as ``"directory/filename"`` but legacy or + application-defined defaults use ``"/directory/filename"`` for + URL-style serving). Storage treats the input as relative to the + upload directory, so we strip a single cosmetic leading slash while + preserving any path-traversal components (which the storage layer + will still reject). + """ + if not path: + return path + if path.startswith("/") and not path.startswith("//"): + return path[1:] + return path + def get_storage(request: Request) -> Any: """Get the storage backend from app.state, or None.""" @@ -24,7 +61,10 @@ def get_storage(request: Request) -> Any: async def _enforce_size_limit( - raw: UploadFile, max_size_mb: float | None, field_name: str, errors: dict[str, list[str]] + raw: UploadFile, + max_size_mb: float | None, + field_name: str, + errors: dict[str, list[str]], ) -> bool: """Enforce the size limit *before* reading the body into RAM. @@ -73,7 +113,7 @@ async def handle_file_field( storage = get_storage(request) field_name = field_meta.name raw = form_data.get(field_name) - + print("file raw", raw, form_data) if isinstance(raw, UploadFile) and raw.filename: # New file uploaded — enforce the size limit before reading content if not await _enforce_size_limit(raw, widget.max_size_mb, field_name, errors): @@ -84,7 +124,8 @@ async def handle_file_field( return try: - path = await storage.save(raw, directory=field_meta.name) + path = await storage.save(raw, directory=_safe_directory_name(field_meta.name)) + print("file path", path) except ValueError as exc: errors[field_name] = [str(exc)] return @@ -94,7 +135,7 @@ async def handle_file_field( old_path = getattr(obj, field_name, None) if old_path: try: - await storage.delete(old_path) + await storage.delete(_normalize_stored_path(old_path)) except ValueError as exc: errors[field_name] = [str(exc)] return @@ -107,7 +148,7 @@ async def handle_file_field( old_path = getattr(obj, field_name, None) if old_path: try: - await storage.delete(old_path) + await storage.delete(_normalize_stored_path(old_path)) except ValueError as exc: errors[field_name] = [str(exc)] return diff --git a/fastapi_admin_kit/views/renderers.py b/fastapi_admin_kit/views/renderers.py index 76bf701..9afcca3 100644 --- a/fastapi_admin_kit/views/renderers.py +++ b/fastapi_admin_kit/views/renderers.py @@ -198,7 +198,7 @@ async def parse( widget = self.registered.get_widget(field_meta.name) if isinstance(widget, _FILE_WIDGET_TYPES): - action = form_data.get(f"_action_{field_meta.name}", "keep") if obj else None + action = form_data.get(f"{field_meta.name}_action", "keep") if obj else None await _handle_file_field( request, widget, diff --git a/fastapi_admin_kit/widgets/inputs.py b/fastapi_admin_kit/widgets/inputs.py index 48f74c3..0012802 100644 --- a/fastapi_admin_kit/widgets/inputs.py +++ b/fastapi_admin_kit/widgets/inputs.py @@ -2,6 +2,7 @@ from __future__ import annotations +import enum import json from datetime import date, datetime from typing import Any @@ -88,19 +89,67 @@ def validate(self, value: Any, field: FieldMeta) -> list[str]: class SelectWidget(Widget): macro_name = "select" - def __init__(self, choices: list[tuple[str, str]] | None = None): + def __init__( + self, + choices: list[tuple[str, str]] | None = None, + enum_class: type | None = None, + ): self.choices = choices or [] + self.enum_class = enum_class + + @staticmethod + def _enum_stored_value(value: enum.Enum) -> Any: + """Return the representation SA actually stores in the DB for this enum. + + SQLAlchemy's ``Enum`` type stores ``.value`` for ``str``-Enum subclasses + (including ``StrEnum``) and ``.name`` for plain ``Enum``. We mirror that + here so the value rendered into the ``