Refactor ctl object endpoint generation

CHANGELOG.md

@@ -20,6 +20,7 @@ The types of changes are:
 ### Changed
 
 - Remove logging within the Celery creation function [#3303](https://github.com/ethyca/fides/pull/3303)
+- Update how generic endpoint generation works [#3304](https://github.com/ethyca/fides/pull/3304)
 - Restrict strack-trace logging when not in Dev mode [#3081](https://github.com/ethyca/fides/pull/3081)
 
 ### Added

src/fides/api/app_setup.py

@@ -14,10 +14,9 @@
 from starlette.middleware.cors import CORSMiddleware
 
 import fides
-from fides.api.ctl import view
 from fides.api.ctl.database.database import configure_db
 from fides.api.ctl.database.seed import create_or_update_parent_user
-from fides.api.ctl.routes import admin, crud, generate, health, system, validate
+from fides.api.ctl.routes import CTL_ROUTER
 from fides.api.ctl.utils.errors import FidesError
 from fides.api.ctl.utils.logger import setup as setup_logging
 from fides.api.ops.api.deps import get_api_session
@@ -46,15 +45,7 @@
 
 VERSION = fides.__version__
 
-ROUTERS = crud.routers + [  # type: ignore[attr-defined]
-    admin.router,
-    generate.router,
-    health.router,
-    validate.router,
-    view.router,
-    system.system_connections_router,
-    system.system_router,
-]
+ROUTERS = [CTL_ROUTER, api_router]
 
 
 def create_fides_app(
@@ -101,7 +92,6 @@ def create_fides_app(
 
     for router in routers:
         fastapi_app.include_router(router)
-    fastapi_app.include_router(api_router)
 
     if security_env == "dev":
         # This removes auth requirements for specific endpoints

src/fides/api/ctl/database/database.py

@@ -12,13 +12,12 @@
 from sqlalchemy_utils.functions import create_database, database_exists
 from sqlalchemy_utils.types.encrypted.encrypted_type import InvalidCiphertextError
 
+from fides.api.ctl.database.seed import load_default_resources, load_samples
+from fides.api.ctl.database.session import async_session
 from fides.api.ctl.utils.errors import get_full_exception_name
 from fides.api.ops.db.base import Base  # type: ignore[attr-defined]
 from fides.core.utils import get_db_engine
 
-from .seed import load_default_resources, load_samples
-from .session import async_session
-
 DatabaseHealth = Literal["healthy", "unhealthy", "needs migration"]
 
 

src/fides/api/ctl/database/seed.py

@@ -9,7 +9,7 @@
 from sqlalchemy.orm import Session
 
 from fides.api.ctl.database.session import sync_session
-from fides.api.ctl.routes.system import upsert_system
+from fides.api.ctl.database.system import upsert_system
 from fides.api.ctl.sql_models import (  # type: ignore[attr-defined]
     Dataset,
     sql_model_map,

src/fides/api/ctl/database/system.py

@@ -0,0 +1,201 @@
+"""
+Functions for interacting with System objects in the database.
+"""
+from typing import Dict, List, Tuple
+
+from fastapi import HTTPException
+from fideslang.models import System as SystemSchema
+from loguru import logger as log
+from sqlalchemy.ext.asyncio import AsyncSession
+from sqlalchemy.orm import Session
+from starlette.status import HTTP_400_BAD_REQUEST, HTTP_404_NOT_FOUND
+
+from fides.api.ctl.database.crud import create_resource, get_resource, update_resource
+from fides.api.ctl.sql_models import (  # type: ignore[attr-defined]
+    DataUse,
+    PrivacyDeclaration,
+    System,
+)
+from fides.api.ctl.utils.errors import NotFoundError
+
+
+def privacy_declaration_logical_id(
+    privacy_declaration: PrivacyDeclaration,
+) -> str:
+    """
+    Helper to standardize a logical 'id' for privacy declarations.
+    As of now, this is based on the `data_use` and the `name` of the declaration, if provided.
+    """
+    return f"{privacy_declaration.data_use}:{privacy_declaration.name or ''}"
+
+
+def get_system(db: Session, fides_key: str) -> System:
+    system = System.get_by(db, field="fides_key", value=fides_key)
+    if system is None:
+        raise HTTPException(
+            status_code=HTTP_404_NOT_FOUND,
+            detail="A valid system must be provided to create or update connections",
+        )
+    return system
+
+
+async def validate_privacy_declarations(db: AsyncSession, system: SystemSchema) -> None:
+    """
+    Ensure that the `PrivacyDeclaration`s on the provided `System` resource are valid:
+     - that they reference valid `DataUse` records
+     - that there are not "duplicate" `PrivacyDeclaration`s as defined by their "logical ID"
+
+    If not, a `400` is raised
+    """
+    logical_ids = set()
+    for privacy_declaration in system.privacy_declarations:
+        try:
+            await get_resource(
+                sql_model=DataUse,
+                fides_key=privacy_declaration.data_use,
+                async_session=db,
+            )
+        except NotFoundError:
+            raise HTTPException(
+                status_code=HTTP_400_BAD_REQUEST,
+                detail=f"Invalid privacy declaration referencing unknown DataUse {privacy_declaration.data_use}",
+            )
+        logical_id = privacy_declaration_logical_id(privacy_declaration)
+        if logical_id in logical_ids:
+            raise HTTPException(
+                status_code=HTTP_400_BAD_REQUEST,
+                detail=f"Duplicate privacy declarations specified with data use {privacy_declaration.data_use}",
+            )
+
+        logical_ids.add(logical_id)
+
+
+async def upsert_system(
+    resources: List[SystemSchema], db: AsyncSession
+) -> Tuple[int, int]:
+    """Helper method to abstract system upsert logic from API code"""
+    inserted = 0
+    updated = 0
+    # first pass to validate privacy declarations before proceeding
+    for resource in resources:
+        await validate_privacy_declarations(db, resource)
+
+    for resource in resources:
+        try:
+            await get_resource(System, resource.fides_key, db)
+        except NotFoundError:
+            log.debug(
+                f"Upsert System with fides_key {resource.fides_key} not found, will create"
+            )
+            await create_system(resource=resource, db=db)
+            inserted += 1
+            continue
+        await update_system(resource=resource, db=db)
+        updated += 1
+    return (inserted, updated)
+
+
+async def upsert_privacy_declarations(
+    db: AsyncSession, resource: SystemSchema, system: System
+) -> None:
+    """Helper to handle the specific upsert logic for privacy declarations"""
+
+    async with db.begin():
+        # map existing declarations by their logical identifier
+        existing_declarations: Dict[str, PrivacyDeclaration] = {
+            privacy_declaration_logical_id(existing_declaration): existing_declaration
+            for existing_declaration in system.privacy_declarations
+        }
+
+        # iterate through declarations specified on the request and upsert
+        # looking for "matching" existing declarations based on data_use and name
+        for privacy_declaration in resource.privacy_declarations:
+            # prepare our 'payload' for either create or update
+            data = privacy_declaration.dict()
+            data["system_id"] = system.id  # include FK back to system
+
+            # if we find matching declaration, remove it from our map
+            if existing_declaration := existing_declarations.pop(
+                privacy_declaration_logical_id(privacy_declaration), None
+            ):
+                # and update existing declaration *in place*
+                existing_declaration.update(db, data=data)
+            else:
+                # otherwise, create a new declaration record
+                PrivacyDeclaration.create(db, data=data)
+
+        # delete any existing privacy declarations that have not been "matched" in the request
+        for existing_declarations in existing_declarations.values():
+            await db.delete(existing_declarations)
+
+
+async def update_system(resource: SystemSchema, db: AsyncSession) -> Dict:
+    """Helper function to share core system update logic for wrapping endpoint functions"""
+    system: System = await get_resource(
+        sql_model=System, fides_key=resource.fides_key, async_session=db
+    )
+
+    # handle the privacy declaration upsert logic
+    try:
+        await upsert_privacy_declarations(db, resource, system)
+    except Exception as e:
+        log.error(
+            f"Error adding privacy declarations, reverting system creation: {str(e)}"
+        )
+        raise e
+
+    delattr(
+        resource, "privacy_declarations"
+    )  # remove the attribute on the system since we've already updated declarations
+
+    # perform any updates on the system resource itself
+    updated_system = await update_resource(System, resource.dict(), db)
+    async with db.begin():
+        await db.refresh(updated_system)
+    return updated_system
+
+
+async def create_system(
+    resource: SystemSchema,
+    db: AsyncSession,
+) -> Dict:
+    """
+    Override `System` create/POST to handle `.privacy_declarations` defined inline,
+    for backward compatibility and ease of use for API users.
+    """
+    await validate_privacy_declarations(db, resource)
+    # copy out the declarations to be stored separately
+    # as they will be processed AFTER the system is added
+    privacy_declarations = resource.privacy_declarations
+
+    # remove the attribute on the system update since the declarations will be created separately
+    delattr(resource, "privacy_declarations")
+
+    # create the system resource using generic creation
+    # the system must be created before the privacy declarations so that it can be referenced
+    created_system = await create_resource(
+        System, resource_dict=resource.dict(), async_session=db
+    )
+
+    privacy_declaration_exception = None
+    try:
+        async with db.begin():
+            # create the specified declarations as records in their own table
+            for privacy_declaration in privacy_declarations:
+                data = privacy_declaration.dict()
+                data["system_id"] = created_system.id  # add FK back to system
+                PrivacyDeclaration.create(
+                    db, data=data
+                )  # create the associated PrivacyDeclaration
+    except Exception as e:
+        log.error(
+            f"Error adding privacy declarations, reverting system creation: {str(privacy_declaration_exception)}"
+        )
+        async with db.begin():
+            await db.delete(created_system)
+        raise e
+
+    async with db.begin():
+        await db.refresh(created_system)
+
+    return created_system

src/fides/api/ctl/routes/__init__.py

@@ -0,0 +1,45 @@
+"""
+Define routes for the ctl-related objects.
+
+All routes should get imported here and added to the CTL_ROUTER
+"""
+from fastapi import APIRouter
+
+from .admin import ADMIN_ROUTER
+from .generate import GENERATE_ROUTER
+from .generic import (
+    DATA_CATEGORY_ROUTER,
+    DATA_QUALIFIER_ROUTER,
+    DATA_SUBJECT_ROUTER,
+    DATA_USE_ROUTER,
+    DATASET_ROUTER,
+    EVALUATION_ROUTER,
+    ORGANIZATION_ROUTER,
+    POLICY_ROUTER,
+    REGISTRY_ROUTER,
+)
+from .health import HEALTH_ROUTER
+from .system import SYSTEM_CONNECTIONS_ROUTER, SYSTEM_ROUTER
+from .validate import VALIDATE_ROUTER
+
+routers = [
+    ADMIN_ROUTER,
+    DATA_CATEGORY_ROUTER,
+    DATA_SUBJECT_ROUTER,
+    DATA_QUALIFIER_ROUTER,
+    DATA_USE_ROUTER,
+    DATASET_ROUTER,
+    EVALUATION_ROUTER,
+    GENERATE_ROUTER,
+    HEALTH_ROUTER,
+    ORGANIZATION_ROUTER,
+    POLICY_ROUTER,
+    REGISTRY_ROUTER,
+    SYSTEM_CONNECTIONS_ROUTER,
+    SYSTEM_ROUTER,
+    VALIDATE_ROUTER,
+]
+
+CTL_ROUTER = APIRouter()
+for router in routers:
+    CTL_ROUTER.include_router(router)

src/fides/api/ctl/routes/admin.py

@@ -11,7 +11,7 @@
 from fides.api.ops.oauth.utils import verify_oauth_client_prod
 from fides.core.config import CONFIG
 
-router = APIRouter(prefix=API_PREFIX, tags=["Admin"])
+ADMIN_ROUTER = APIRouter(prefix=API_PREFIX, tags=["Admin"])
 
 
 class DBActions(str, Enum):
@@ -20,7 +20,7 @@ class DBActions(str, Enum):
     reset = "reset"
 
 
-@router.post(
+@ADMIN_ROUTER.post(
     "/admin/db/{action}",
     tags=["Admin"],
     dependencies=[