Refactor to support mass insert/update

app.py

@@ -117,6 +117,9 @@ def create_app() -> Flask:
     app.config["JWT_ACCESS_TOKEN_EXPIRES"] = timedelta(minutes=15)
     app.config["JWT_REFRESH_TOKEN_EXPIRES"] = timedelta(days=30)
 
+    # Other configuration settings
+    app.config["MAX_CONTENT_LENGTH"] = 16 * 1024 * 1024
+
     app.secret_key = get_flask_app_cookie_encryption_key()
     app.wsgi_app = ReverseProxied(app.wsgi_app)
     CORS(app)

middleware/schema_and_dto_logic/dynamic_logic/dynamic_schema_request_content_population.py

@@ -85,14 +85,7 @@ def _get_data_from_sources(fields: dict, schema: SchemaTypes) -> SourceDataInfo:
         metadata = field_value.metadata
         source: SourceMappingEnum = _get_required_argument("source", metadata, schema)
         if isinstance(field_value, marshmallow.fields.Nested):
-            if source != SourceMappingEnum.JSON:
-                raise InvalidSourceMappingError(
-                    "Nested fields can only be populated from JSON sources"
-                )
-            if "nested_dto_class" not in metadata:
-                raise InvalidSourceMappingError(
-                    "Nested fields must have a 'nested_dto_class' metadata"
-                )
+            _check_for_errors(metadata, source)
             nested_dto_class = metadata["nested_dto_class"]
             nested_dto_info_list.append(
                 NestedDTOInfo(key=field_name, class_=nested_dto_class)
@@ -105,6 +98,17 @@ def _get_data_from_sources(fields: dict, schema: SchemaTypes) -> SourceDataInfo:
     return SourceDataInfo(data=data, nested_dto_info_list=nested_dto_info_list)
 
 
+def _check_for_errors(metadata: dict, source: SourceMappingEnum):
+    if source != SourceMappingEnum.JSON:
+        raise InvalidSourceMappingError(
+            "Nested fields can only be populated from JSON sources"
+        )
+    if "nested_dto_class" not in metadata:
+        raise InvalidSourceMappingError(
+            "Nested fields must have a 'nested_dto_class' metadata"
+        )
+
+
 def _apply_transformation_functions_to_dict(fields: dict, intermediate_data: dict):
     """
     Apply transformation functions to the data,

middleware/schema_and_dto_logic/util.py

@@ -23,13 +23,16 @@ def _get_required_argument(
 
 
 def _get_source_getting_function(source: SourceMappingEnum) -> Callable:
+    # TODO: Consider moving this to separate map variable,
+    #        rather than define within the function
     source_mapping: dict[SourceMappingEnum, Callable] = {
         SourceMappingEnum.QUERY_ARGS: request.args.get,
         SourceMappingEnum.FORM: request.form.get,
         SourceMappingEnum.JSON: lambda key: (
             request.json.get(key) if request.json else None
         ),
         SourceMappingEnum.PATH: request.view_args.get,
+        SourceMappingEnum.FILE: request.files.get,
     }
     return source_mapping[source]
 

resources/Login.py

@@ -5,17 +5,13 @@
 from middleware.decorators import endpoint_info
 from middleware.primary_resource_logic.login_queries import try_logging_in
 from resources.endpoint_schema_config import SchemaConfigs
-from resources.resource_helpers import create_jwt_tokens_model, ResponseInfo
-from middleware.schema_and_dto_logic.dynamic_logic.model_helpers_with_schemas import (
-    create_user_model,
-)
+from resources.resource_helpers import ResponseInfo
+
 from utilities.namespace import create_namespace
 
 from resources.PsycopgResource import PsycopgResource, handle_exceptions
 
 namespace_login = create_namespace()
-user_model = create_user_model(namespace_login)
-jwt_tokens_model = create_jwt_tokens_model(namespace_login)
 
 
 @namespace_login.route("/login")

resources/resource_helpers.py

@@ -106,103 +106,6 @@ def create_response_dictionary(
     }
 
 
-def create_jwt_tokens_model(namespace: Namespace) -> Model:
-    return namespace.model(
-        "JWTTokens",
-        {
-            "access_token": fields.String(
-                required=True, description="The access token of the user"
-            ),
-            "refresh_token": fields.String(
-                required=True, description="The refresh token of the user"
-            ),
-        },
-    )
-
-
-def create_search_model(namespace: Namespace) -> Model:
-    search_result_inner_model = namespace.model(
-        "SearchResultInner",
-        {
-            "airtable_uid": fields.String(
-                required=True, description="Airtable UID of the record"
-            ),
-            "agency_name": fields.String(description="Name of the agency"),
-            "municipality": fields.String(description="Name of the municipality"),
-            "state_iso": fields.String(description="ISO code of the state"),
-            "data_source_name": fields.String(description="Name of the data source"),
-            "description": fields.String(description="Description of the record"),
-            "record_type": fields.String(description="Type of the record"),
-            "source_url": fields.String(description="URL of the data source"),
-            "record_format": fields.String(description="Format of the record"),
-            "coverage_start": fields.String(description="Coverage start date"),
-            "coverage_end": fields.String(description="Coverage end date"),
-            "agency_supplied": fields.String(
-                description="If the record is supplied by the agency"
-            ),
-            "jurisdiction_type": fields.String(
-                description="Type of jursidiction for agency"
-            ),
-        },
-    )
-
-    search_result_inner_wrapper_model = namespace.model(
-        "SearchResultInnerWrapper",
-        {
-            "count": fields.Integer(
-                required=True,
-                description="Count of SearchResultInnerWrapper items",
-                attribute="count",
-            ),
-            "results": fields.List(
-                fields.Nested(
-                    search_result_inner_model,
-                    description="List of results for the given jurisdiction",
-                )
-            ),
-        },
-    )
-
-    search_result_jurisdictions_wrapper_model = namespace.model(
-        name="SearchResultJurisdictionsWrapper",
-        model={
-            "federal": fields.Nested(
-                search_result_inner_wrapper_model,
-                description="Results for the federal jurisdiction",
-            ),
-            "state": fields.Nested(
-                search_result_inner_wrapper_model,
-                description="Results for the state jurisdiction",
-            ),
-            "county": fields.Nested(
-                search_result_inner_wrapper_model,
-                description="Results for the county jurisdiction",
-            ),
-            "locality": fields.Nested(
-                search_result_inner_wrapper_model,
-                description="Results for the locality jurisdiction",
-            ),
-        },
-    )
-
-    search_result_outer_model = namespace.model(
-        "SearchResultOuter",
-        {
-            "count": fields.Integer(
-                required=True,
-                description="Count of SearchResultInner items",
-                attribute="count",
-            ),
-            "data": fields.Nested(
-                attribute="data",
-                model=search_result_jurisdictions_wrapper_model,
-            ),
-        },
-    )
-
-    return search_result_outer_model
-
-
 def column_permissions_description(
     head_description: str,
     column_permissions_str_table: str,

utilities/enums.py

@@ -24,6 +24,7 @@ class SourceMappingEnum(Enum):
     FORM = "form"
     JSON = "json"
     PATH = "path"
+    FILE = "file"
 
 
 class ParserLocation(Enum):