Fix Fideslang Typing

.github/workflows/backend_checks.yml

@@ -79,6 +79,7 @@ jobs:
           [
             "isort",
             "black",
+            "mypy",
             "pylint",
             "xenon",
             "check_install",

noxfiles/docker_nox.py

@@ -2,6 +2,7 @@
 from typing import Callable, Dict, List, Optional, Tuple
 
 import nox
+
 from constants_nox import (
     DEV_TAG_SUFFIX,
     IMAGE,

pyproject.toml

@@ -43,7 +43,6 @@ module = [
   "dask.*",
   "deepdiff.*",
   "defusedxml.ElementTree.*",
-  "fideslang.*",
   "fideslog.*",
   "firebase_admin.*",
   "google.api_core.*",

src/fides/api/alembic/migrations/versions/f17f92237383_disable_inactive_notices.py

@@ -13,7 +13,6 @@
 from sqlalchemy.engine import Connection, ResultProxy
 from sqlalchemy.sql.elements import TextClause
 
-
 # revision identifiers, used by Alembic.
 from fides.api.models.sql_models import DataUse
 

src/fides/api/api/v1/endpoints/generic.py

@@ -2,7 +2,7 @@
 This module generates all of the routers for the boilerplate/generic
 objects that don't require any extra logic.
 """
-from fideslang import Dataset, Evaluation, Organization, Policy, Registry
+from fideslang.models import Dataset, Evaluation, Organization, Policy, Registry
 
 from fides.api.schemas.taxonomy_extensions import (
     DataCategory,

src/fides/api/api/v1/endpoints/privacy_experience_endpoints.py

@@ -67,20 +67,20 @@ def _filter_experiences_by_region_or_country(
     if not region:
         return experience_query
 
-    region: str = escape(region).replace("-", "_").lower()
-    country: str = region.split("_")[0]
+    formatted_region: str = escape(region).replace("-", "_").lower()
+    country: str = formatted_region.split("_")[0]
 
     overlay: Optional[
         PrivacyExperience
     ] = PrivacyExperience.get_experience_by_region_and_component(
-        db, region, ComponentType.overlay
+        db, formatted_region, ComponentType.overlay
     ) or PrivacyExperience.get_experience_by_region_and_component(
         db, country, ComponentType.overlay
     )
     privacy_center: Optional[
         PrivacyExperience
     ] = PrivacyExperience.get_experience_by_region_and_component(
-        db, region, ComponentType.privacy_center
+        db, formatted_region, ComponentType.privacy_center
     ) or PrivacyExperience.get_experience_by_region_and_component(
         db, country, ComponentType.privacy_center
     )

src/fides/api/api/v1/endpoints/router_factory.py

@@ -8,7 +8,8 @@
 from typing import Dict, List
 
 from fastapi import Depends, HTTPException, Response, Security, status
-from fideslang import Dataset, FidesModelType
+from fideslang import FidesModelType
+from fideslang.models import Dataset
 from fideslang.validation import FidesKey
 from pydantic import ValidationError as PydanticValidationError
 from sqlalchemy.ext.asyncio import AsyncSession

src/fides/api/db/samples.py

@@ -8,7 +8,7 @@
 
 import yaml
 from expandvars import expandvars  # type: ignore
-from fideslang import Taxonomy
+from fideslang.models import Taxonomy
 from fideslang.validation import FidesKey
 
 # DEFER: This can be changed to importlib.resources once we drop support for Python 3.8

src/fides/api/db/seed.py

@@ -3,7 +3,7 @@
 """
 from typing import Dict, List, Optional
 
-from fideslang import DEFAULT_TAXONOMY
+from fideslang.default_taxonomy import DEFAULT_TAXONOMY
 from loguru import logger as log
 from sqlalchemy.ext.asyncio import AsyncSession
 from sqlalchemy.orm import Session

src/fides/api/db/system.py

@@ -146,20 +146,27 @@ async def upsert_privacy_declarations(
 
 async def upsert_cookies(
     async_session: AsyncSession,
-    cookies: Optional[List[Dict]],  # CookieSchema
+    cookies: Optional[List[Dict]],
     privacy_declaration: PrivacyDeclaration,
     system: System,
 ) -> None:
-    """Upsert cookies for the given privacy declaration: retrieve cookies by name/system/privacy declaration
+    """
+    Upsert cookies for the given privacy declaration: retrieve cookies
+    by name/system/privacy declaration.
+
     Remove any existing cookies that aren't specified here.
     """
-    cookie_list: List[CookieSchema] = cookies or []
-    for cookie_data in cookie_list:
+
+    parsed_cookies = (
+        [CookieSchema.parse_obj(cookie) for cookie in cookies] if cookies else []
+    )
+
+    for cookie_data in parsed_cookies:
         # Check if cookie exists for this name/system/privacy declaration
         result = await async_session.execute(
             select(Cookies).where(
                 and_(
-                    Cookies.name == cookie_data["name"],
+                    Cookies.name == cookie_data.name,
                     Cookies.system_id == system.id,
                     Cookies.privacy_declaration_id == privacy_declaration.id,
                 )
@@ -168,16 +175,16 @@ async def upsert_cookies(
         row: Optional[Cookies] = result.scalars().first()
         if row:
             await async_session.execute(
-                update(Cookies).where(Cookies.id == row.id).values(cookie_data)
+                update(Cookies).where(Cookies.id == row.id).values(cookie_data.dict())
             )
 
         else:
             await async_session.execute(
                 insert(Cookies).values(
                     {
-                        "name": cookie_data.get("name"),
-                        "path": cookie_data.get("path"),
-                        "domain": cookie_data.get("domain"),
+                        "name": cookie_data.name,
+                        "path": cookie_data.path,
+                        "domain": cookie_data.domain,
                         "privacy_declaration_id": privacy_declaration.id,
                         "system_id": system.id,
                     }
@@ -188,7 +195,7 @@ async def upsert_cookies(
     delete_result = await async_session.execute(
         select(Cookies).where(
             and_(
-                Cookies.name.notin_([cookie["name"] for cookie in cookie_list]),
+                Cookies.name.notin_([cookie.name for cookie in parsed_cookies]),
                 Cookies.system_id == system.id,
                 Cookies.privacy_declaration_id == privacy_declaration.id,
             )

src/fides/api/models/datasetconfig.py

@@ -168,7 +168,7 @@
 
         return dataset_graph
 
-    def get_dataset_with_stubbed_collection(self) -> Dataset:
+    def get_dataset_with_stubbed_collection(self) -> GraphDataset:
         """
         Return a Dataset with a single mock Collection for use in building a graph
         where we only want one node per dataset, instead of one node per collection.  Note that
@@ -179,7 +179,7 @@
         is that this single node represents the overall Dataset, and will execute Dataset-level requests,
         not Collection-level requests.
         """
-        dataset_graph: Dataset = self.get_graph()
+        dataset_graph = self.get_graph()
         stubbed_collection = Collection(name=dataset_graph.name, fields=[], after=set())
 
         dataset_graph.collections = [stubbed_collection]
@@ -256,7 +256,7 @@
         data_categories=field.data_categories,
         identity=identity,
         data_type_name=data_type_name,  # type: ignore
-        references=references,
+        references=references,  # type: ignore
         is_pk=is_pk,
         length=length,
         is_array=is_array,

src/fides/api/models/policy.py

@@ -2,7 +2,7 @@
 from enum import Enum as EnumType
 from typing import Any, Dict, List, Optional, Tuple, Union
 
-from fideslang import DEFAULT_TAXONOMY
+from fideslang.default_taxonomy import DEFAULT_TAXONOMY
 from fideslang.models import DataCategory as FideslangDataCategory
 from fideslang.validation import FidesKey
 from sqlalchemy import Column

src/fides/api/models/privacy_notice.py

@@ -202,8 +202,7 @@ def generate_notice_key(cls, name: Optional[str]) -> FidesKey:
         if not isinstance(name, str):
             raise Exception("Privacy notice keys must be generated from a string.")
         notice_key: str = re.sub(r"\s+", "_", name.lower().strip())
-        FidesKey.validate(notice_key)
-        return notice_key
+        return FidesKey(FidesKey.validate(notice_key))
 
     def dry_update(self, *, data: dict[str, Any]) -> FidesBase:
         """

src/fides/api/models/sql_models.py

@@ -553,7 +553,7 @@ class SystemScans(Base):
 
 models_with_default_field = [
     sql_model
-    for _, sql_model in sql_model_map.items()
+    for sql_model in sql_model_map.values()
     if hasattr(sql_model, "is_default")
 ]
 

src/fides/api/oauth/system_manager_oauth_util.py

@@ -78,7 +78,7 @@ async def verify_oauth_client_for_system_from_request_body(
     authorization: str = Security(oauth2_scheme),
     db: Session = Depends(get_db),
     system_auth_data: SystemAuthContainer = Depends(_get_system_from_request_body),
-) -> SystemSchema:
+) -> Union[str, System]:
     """
     Verifies that the access token provided in the authorization header contains the necessary scopes to be
     able to access the System found in the *request body*
@@ -88,20 +88,22 @@ async def verify_oauth_client_for_system_from_request_body(
 
     Yields a 403 forbidden error if not.
     """
-    return has_system_permissions(
+
+    system = has_system_permissions(
         system_auth_data=system_auth_data,
         authorization=authorization,
         security_scopes=security_scopes,
         db=db,
     )
+    return system
 
 
 async def verify_oauth_client_for_system_from_fides_key(
     security_scopes: SecurityScopes,
     authorization: str = Security(oauth2_scheme),
     db: Session = Depends(get_db),
     system_auth_data: SystemAuthContainer = Depends(_get_system_from_fides_key),
-) -> str:
+) -> Union[str, System]:
     """
     Verifies that the access token provided in the authorization header contains the necessary scopes to be
     able to access the System from the given *fides_key* in the path parameter.
@@ -111,20 +113,21 @@ async def verify_oauth_client_for_system_from_fides_key(
 
     Yields a 403 forbidden error if not.
     """
-    return has_system_permissions(
+    system = has_system_permissions(
         system_auth_data=system_auth_data,
         authorization=authorization,
         security_scopes=security_scopes,
         db=db,
     )
+    return system
 
 
 def has_system_permissions(
     system_auth_data: SystemAuthContainer,
     authorization: str,
     security_scopes: SecurityScopes,
     db: Session,
-) -> Union[str, SystemSchema]:
+) -> Union[str, System]:
     """
     Helper method that verifies that the token has the proper permissions to access the system(s).
 

src/fides/api/schemas/connection_configuration/connection_secrets_saas.py

@@ -1,7 +1,7 @@
 import abc
 from typing import Any, Dict, List, Type
 
-from fideslang import FidesDatasetReference
+from fideslang.models import FidesDatasetReference
 from pydantic import BaseModel, Extra, Field, PrivateAttr, create_model, root_validator
 from pydantic.fields import FieldInfo
 from sqlalchemy.orm import Session

src/fides/api/schemas/dataset.py

@@ -1,4 +1,4 @@
-from typing import Any, List, Optional, Type
+from typing import Any, List, Optional, Sequence, Type
 
 from fideslang.models import Dataset, DatasetCollection, DatasetField
 from fideslang.validation import FidesKey
@@ -25,7 +25,9 @@ def validate_data_categories_against_db(
         logger.info(
             "No data categories in the database: reverting to default data categories."
         )
-        defined_data_categories = list(DefaultTaxonomyDataCategories.__members__.keys())
+        defined_data_categories = [
+            FidesKey(key) for key in DefaultTaxonomyDataCategories.__members__.keys()
+        ]
 
     class DataCategoryValidationMixin(BaseModel):
         @validator("data_categories", check_fields=False, allow_reuse=True)
@@ -36,17 +38,17 @@ def valid_data_categories(
             return _valid_data_categories(v, defined_data_categories)
 
     class FieldDataCategoryValidation(DatasetField, DataCategoryValidationMixin):
-        fields: Optional[List["FieldDataCategoryValidation"]]
+        fields: Optional[List["FieldDataCategoryValidation"]]  # type: ignore[assignment]
 
     FieldDataCategoryValidation.update_forward_refs()
 
     class CollectionDataCategoryValidation(
         DatasetCollection, DataCategoryValidationMixin
     ):
-        fields: List[FieldDataCategoryValidation] = []
+        fields: Sequence[FieldDataCategoryValidation] = []  # type: ignore[assignment]
 
     class DatasetDataCategoryValidation(Dataset, DataCategoryValidationMixin):
-        collections: List[CollectionDataCategoryValidation]
+        collections: Sequence[CollectionDataCategoryValidation]  # type: ignore[assignment]
 
     DatasetDataCategoryValidation(**dataset.dict())
 

src/fides/api/schemas/messaging/messaging.py

@@ -3,7 +3,7 @@
 from enum import Enum
 from typing import Any, Dict, List, Optional, Tuple, Type, Union
 
-from fideslang import DEFAULT_TAXONOMY
+from fideslang.default_taxonomy import DEFAULT_TAXONOMY
 from fideslang.validation import FidesKey
 from pydantic import BaseModel, Extra, root_validator
 
@@ -126,7 +126,7 @@ def transform_data_use_format(cls, values: Dict[str, Any]) -> Dict[str, Any]:
             preference.data_use = next(
                 (
                     data_use.name
-                    for data_use in DEFAULT_TAXONOMY.data_use
+                    for data_use in DEFAULT_TAXONOMY.data_use or []
                     if data_use.fides_key == preference.data_use
                 ),
                 preference.data_use,

src/fides/api/schemas/system.py

@@ -1,5 +1,5 @@
 from datetime import datetime
-from typing import Any, Dict, List, Optional
+from typing import Any, Dict, List, Optional, Sequence
 
 from fideslang.models import Cookies, PrivacyDeclaration, System
 from pydantic import Field
@@ -24,7 +24,7 @@ class SystemResponse(System):
     """Extension of base pydantic model to include additional fields like `privacy_declarations`, connection_config fields
     and cookies in responses"""
 
-    privacy_declarations: List[PrivacyDeclarationResponse] = Field(
+    privacy_declarations: Sequence[PrivacyDeclarationResponse] = Field(  # type: ignore[assignment]
         description=PrivacyDeclarationResponse.__doc__,
     )
 

src/fides/api/service/privacy_request/request_runner_service.py

@@ -502,7 +502,7 @@ def build_consent_dataset_graph(datasets: List[DatasetConfig]) -> DatasetGraph:
             and saas_config.get("consent_requests")
         ):
             consent_datasets.append(
-                dataset_config.get_dataset_with_stubbed_collection()
+                dataset_config.get_dataset_with_stubbed_collection()  # type: ignore[arg-type, assignment]
             )
 
     return DatasetGraph(*consent_datasets)

src/fides/api/util/connection_util.py

@@ -1,4 +1,4 @@
-from typing import List, Optional, Union
+from typing import List, Optional
 
 from fastapi import Depends, HTTPException
 from fideslang.validation import FidesKey
@@ -325,7 +325,7 @@ def connection_status(
 
     connector = get_connector(connection_config)
     try:
-        status: Union[ConnectionTestStatus, None] = connector.test_connection()
+        status: Optional[ConnectionTestStatus] = connector.test_connection()
 
     except (ConnectionException, ClientUnsuccessfulException) as exc:
         logger.warning(

src/fides/api/util/data_category.py

@@ -1,7 +1,7 @@
 from enum import Enum as EnumType
 from typing import List, Type
 
-from fideslang import DEFAULT_TAXONOMY
+from fideslang.default_taxonomy import DEFAULT_TAXONOMY
 from fideslang.validation import FidesKey
 from sqlalchemy.orm import Session