Add Starlette Admin interface for Location management (MS Access replacement)

.env.example

@@ -3,6 +3,20 @@ DB_DRIVER=postgres
 POSTGRES_USER=admin
 POSTGRES_PASSWORD=password
 POSTGRES_DB=
+POSTGRES_HOST=localhost
+POSTGRES_PORT=5432
+
+# Connection pool configuration for parallel transfers
+# pool_size: number of persistent connections to maintain
+# max_overflow: additional connections allowed during peak usage
+DB_POOL_SIZE=10
+DB_MAX_OVERFLOW=20
+
+# Transfer configuration
+# Enable parallel transfers (default: true)
+TRANSFER_PARALLEL=1
+# Limit number of records per transfer type (for testing)
+# TRANSFER_LIMIT=1000
 
 # asset storage
 GCS_BUCKET_NAME=
@@ -19,3 +33,6 @@ AUTHENTIK_URL=
 AUTHENTIK_CLIENT_ID=
 AUTHENTIK_AUTHORIZE_URL=
 AUTHENTIK_TOKEN_URL=
+
+# middleware
+SESSION_SECRET_KEY=your_secret_key_here

.github/workflows/CD_production.yml

@@ -58,6 +58,7 @@ jobs:
               AUTHENTIK_CLIENT_ID: "${{ vars.AUTHENTIK_CLIENT_ID }}"
               AUTHENTIK_AUTHORIZE_URL: "${{ vars.AUTHENTIK_AUTHORIZE_URL }}"
               AUTHENTIK_TOKEN_URL: "${{ vars.AUTHENTIK_TOKEN_URL }}"
+              SESSION_SECRET_KEY: "${{ secrets.SESSION_SECRET_KEY }}"
             EOF
 
         - name: Deploy to Google Cloud

.github/workflows/CD_staging.yml

@@ -33,6 +33,7 @@ jobs:
               --output-file requirements.txt
 
         - name: Authenticate to Google Cloud
+          id: gcloud-auth
           uses: 'google-github-actions/auth@v2'
           with:
             credentials_json: ${{ secrets.CLOUD_SQL_SERVICE_ACCOUNT_KEY }}
@@ -64,6 +65,16 @@ jobs:
           with:
             credentials_json: ${{ secrets.CLOUD_DEPLOY_SERVICE_ACCOUNT_KEY }}
 
+        - name: Run Alembic migrations on staging database
+          env:
+            DB_DRIVER: "cloudsql"
+            CLOUD_SQL_INSTANCE_NAME: "${{ secrets.CLOUD_SQL_INSTANCE_NAME }}"
+            CLOUD_SQL_DATABASE: "${{ vars.CLOUD_SQL_DATABASE }}"
+            CLOUD_SQL_USER: "${{ secrets.CLOUD_SQL_USER }}"
+            CLOUD_SQL_PASSWORD: "${{ secrets.CLOUD_SQL_PASSWORD }}"
+          run: |
+            uv run alembic upgrade head
+
         # Uses Google Cloud Secret Manager to store secret credentials
         - name: Create app.yaml
           run: |
@@ -89,6 +100,7 @@ jobs:
               AUTHENTIK_CLIENT_ID: "${{ vars.AUTHENTIK_CLIENT_ID }}"
               AUTHENTIK_AUTHORIZE_URL: "${{ vars.AUTHENTIK_AUTHORIZE_URL }}"
               AUTHENTIK_TOKEN_URL: "${{ vars.AUTHENTIK_TOKEN_URL }}"
+              SESSION_SECRET_KEY: "${{ secrets.SESSION_SECRET_KEY }}"
             EOF
 
         - name: Deploy to Google Cloud

.github/workflows/tests.yml

@@ -23,6 +23,7 @@ jobs:
       POSTGRES_PASSWORD: postgres
       DB_DRIVER: postgres
       BASE_URL: http://localhost:8000
+      SESSION_SECRET_KEY: supersecretkeyforunittests
 
     services:
       postgis:

admin/__init__.py

@@ -0,0 +1,23 @@
+# ===============================================================================
+# Copyright 2025
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+# ===============================================================================
+"""
+Starlette Admin package for NMSampleLocations.
+
+Provides web-based administrative interface for managing database records.
+"""
+from admin.config import create_admin
+
+__all__ = ["create_admin"]

admin/auth.py

@@ -0,0 +1,264 @@
+# ===============================================================================
+# Copyright 2025
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+# ===============================================================================
+"""
+Admin authentication provider integrating with existing Authentik OIDC auth.
+
+This module provides a Starlette Admin AuthProvider that integrates with the
+existing Authentik-based authentication system used by the NMSampleLocations API.
+"""
+import os
+from typing import Optional
+
+from dataclasses import dataclass
+from typing import List
+
+from starlette.requests import Request
+from starlette.responses import RedirectResponse
+from starlette_admin.auth import AdminUser, AuthProvider
+from starlette_admin.exceptions import LoginFailed
+
+from core.permissions import _get_token_payload, verify_token
+
+
+@dataclass
+class AdminUserWithRoles(AdminUser):
+    """Extended AdminUser with roles for RBAC."""
+
+    roles: List[str] = None
+
+    def __post_init__(self):
+        if self.roles is None:
+            self.roles = []
+
+
+class NMSampleLocationsAuthProvider(AuthProvider):
+    """
+    Custom auth provider that integrates with existing Authentik OIDC authentication.
+
+    Reuses the existing authentication infrastructure from core.permissions module.
+
+    For MS Access users: This replaces Access file-level security with user-level
+    authentication. Each user logs in with their Authentik credentials and gets
+    assigned roles (Admin, Editor, Viewer) which control what they can do in the
+    admin interface.
+    """
+
+    async def is_authenticated(self, request: Request) -> bool:
+        """
+        Check if user is authenticated by verifying their JWT token.
+
+        This method is called on every admin page request to determine if the
+        user should be allowed access.
+
+        Returns:
+            bool: True if user has a valid JWT token, False otherwise
+        """
+        # Check if authentication is disabled (development mode only)
+        if int(os.environ.get("AUTHENTIK_DISABLE_AUTHENTICATION", 0)):
+            from core.settings import settings
+
+            if settings.mode != "production":
+                # Allow unauthenticated access in development mode
+                request.state.user = AdminUserWithRoles(
+                    username="dev_user", roles=["admin"]
+                )
+                return True
+
+        try:
+            # Try to get token from Authorization header
+            authorization = request.headers.get("Authorization")
+            if not authorization:
+                # Try to get token from session/cookie
+                token = request.session.get("token")
+                if not token:
+                    return False
+            else:
+                # Extract token from "Bearer <token>" format
+                token = (
+                    authorization.split(" ")[1]
+                    if " " in authorization
+                    else authorization
+                )
+
+            # Verify token using existing authentication system
+            is_valid = verify_token(token, scope=None, permissions=None)
+
+            if is_valid:
+                # Store user in request state for later access
+                request.state.user = self._create_admin_user_from_token(token)
+
+            return is_valid
+        except Exception:
+            return False
+
+    def _create_admin_user_from_token(self, token: str) -> Optional[AdminUser]:
+        """
+        Extract user information from JWT token and create AdminUser instance.
+
+        Args:
+            token: JWT access token from Authentik
+
+        Returns:
+            AdminUser instance with username and roles, or None if token invalid
+        """
+        try:
+            # Decode JWT payload
+            payload = _get_token_payload(token)
+
+            # Extract user information from JWT claims
+            username = (
+                payload.get("preferred_username")
+                or payload.get("email")
+                or payload.get("sub")
+            )
+            email = payload.get("email")
+            groups = payload.get("groups", [])
+
+            # Map Authentik groups to admin roles
+            roles = []
+
+            # Standard roles
+            if "Admin" in groups:
+                roles.append("admin")
+            if "Editor" in groups:
+                roles.append("editor")
+            if "Viewer" in groups:
+                roles.append("viewer")
+
+            # AMP-specific roles (for AMPAPI-related data)
+            if "AMPAdmin" in groups:
+                roles.append("amp_admin")
+            if "AMPEditor" in groups:
+                roles.append("amp_editor")
+            if "AMPViewer" in groups:
+                roles.append("amp_viewer")
+
+            # Lexicon-specific roles
+            if "LexiconAdmin" in groups:
+                roles.append("lexicon_admin")
+            if "LexiconEditor" in groups:
+                roles.append("lexicon_editor")
+
+            return AdminUserWithRoles(
+                username=username,
+                photo_url=None,  # Could add user avatar URL from OIDC if available
+                roles=roles,
+            )
+        except Exception:
+            return None
+
+    def get_admin_user(self, request: Request) -> Optional[AdminUser]:
+        """
+        Get the current admin user from the request.
+
+        This method is called by Starlette Admin to get user information for
+        display in the UI and permission checks.
+
+        Returns:
+            AdminUser instance with username and roles, or None if not authenticated
+        """
+        # Check if user is already stored in request state
+        if hasattr(request.state, "user"):
+            return request.state.user
+
+        try:
+            # Get token from request
+            authorization = request.headers.get("Authorization")
+            if not authorization:
+                token = request.session.get("token")
+                if not token:
+                    return None
+            else:
+                token = (
+                    authorization.split(" ")[1]
+                    if " " in authorization
+                    else authorization
+                )
+
+            # Create AdminUser from token
+            admin_user = self._create_admin_user_from_token(token)
+
+            # Store in request state for future calls
+            if admin_user:
+                request.state.user = admin_user
+
+            return admin_user
+        except Exception:
+            return None
+
+    async def login(self, *args, **kwargs) -> RedirectResponse:
+        """
+        Redirect to Authentik OIDC login page.
+
+        Note: Starlette Admin will show a login form, but we ignore the username/password
+        and redirect to Authentik OAuth flow instead.
+
+        Args:
+            request: Starlette request object (extracted from args/kwargs)
+            *args/**kwargs: Ignored, kept for compatibility with different
+            Starlette Admin login call signatures
+
+        Returns:
+            RedirectResponse to Authentik authorization endpoint
+        """
+        # Starlette Admin has changed the AuthProvider.login signature across versions.
+        # Accept *args/**kwargs and extract the Request to stay compatible whether
+        # it calls login(request, data, ...) or login(username, password, remember_me, request).
+        request: Optional[Request] = kwargs.get("request")
+        if request is None:
+            for arg in args:
+                if isinstance(arg, Request):
+                    request = arg
+                    break
+
+        if request is None:
+            raise LoginFailed("Unable to determine login request context.")
+
+        # Redirect to Authentik OAuth authorization endpoint
+        authentik_authorize_url = os.environ.get("AUTHENTIK_AUTHORIZE_URL")
+        if not authentik_authorize_url:
+            raise LoginFailed(
+                "Authentik authentication is not configured. Please set AUTHENTIK_AUTHORIZE_URL environment variable."
+            )
+
+        # Store original URL to redirect back after login
+        original_url = str(request.url_for("admin:index"))
+        request.session["auth_redirect"] = original_url
+
+        # Redirect to Authentik login
+        return RedirectResponse(url=authentik_authorize_url, status_code=302)
+
+    async def logout(self, request: Request) -> RedirectResponse:
+        """
+        Handle logout by clearing session and redirecting.
+
+        Args:
+            request: Starlette request object
+
+        Returns:
+            RedirectResponse to home page
+        """
+        # Clear session tokens
+        request.session.pop("token", None)
+        request.session.pop("auth_redirect", None)
+
+        # Clear user from request state
+        if hasattr(request.state, "user"):
+            delattr(request.state, "user")
+
+        # Redirect to home page
+        # TODO: Consider redirecting to Authentik logout endpoint to fully log out
+        return RedirectResponse(url="/", status_code=302)