add EncondingSchema enum and start implementation on default schema representations as a concept in the library

Author: tipatterson-dev

conSys4Py/constants.py

@@ -90,3 +90,19 @@ class APIResourceTypes(Enum):
     SYSTEM = "System"
     SYSTEM_EVENT = "SystemEvent"
     SYSTEM_HISTORY = "SystemHistory"
+
+
+class EncodingSchema(Enum):
+    """
+    Defines the encoding formats
+    """
+    JSON = "application/json"
+    XML = "application/xml"
+    SWE_XML = "application/swe+xml"
+    SWE_JSON = "application/swe+json"
+    SWE_CSV = "application/swe+csv"
+    SWE_BINARY = "application/swe+binary"
+    SWE_TEXT = "application/swe+text"
+    GEO_JSON = "application/geo+json"
+    SML_JSON = "application/sml+json"
+    OM_JSON = "application/om+json"

conSys4Py/core/default_api_helpers.py

@@ -1,7 +1,11 @@
+from __future__ import annotations
+
 from abc import ABC
 from dataclasses import dataclass
 
-from conSys4Py import APIResourceTypes, APITerms
+from pydantic import BaseModel, Field
+
+from conSys4Py import APIResourceTypes, APITerms, EncodingSchema
 from conSys4Py.con_sys_api import ConnectedSystemAPIRequest
 
 
@@ -73,30 +77,35 @@ def resource_type_to_endpoint(res_type: APIResourceTypes, parent_type: APIResour
 @dataclass
 class APIHelper(ABC):
     server_url = None
-    api_endpoint = "/api"
+    api_root = "/api"
     username = None
     password = None
     user_auth = False
 
     def create_resource(self, res_type: APIResourceTypes, json_data: any, parent_res_id: str = None,
-                        from_collection: bool = False):
+                        from_collection: bool = False, url_endpoint: str = None):
         """
         Creates a resource of the given type with the given data, will attempt to create a sub-resource if parent_res_id
         is provided.
         :param res_type:
         :param json_data:
         :param parent_res_id:
         :param from_collection:
+        :param url_endpoint: If given, will override the default URL construction. Should contain the endpoint past the API root.
         :return:
         """
-        url = self.resource_url_resolver(res_type, None, parent_res_id, from_collection)
+
+        if url_endpoint is None:
+            url = self.resource_url_resolver(res_type, None, parent_res_id, from_collection)
+        else:
+            url = f'{self.server_url}/{self.api_root}/{url_endpoint}'
         api_request = ConnectedSystemAPIRequest(url=url, request_method='POST', auth=self.get_helper_auth(),
                                                 body=json_data)
         return api_request.make_request()
 
     def retrieve_resource(self, res_type: APIResourceTypes, res_id: str, parent_res_id: str = None,
                           from_collection: bool = False,
-                          collection_id: str = None):
+                          collection_id: str = None, url_endpoint: str = None):
         """
         Retrieves a resource or list of resources if no res_id is provided, will attempt to retrieve a sub-resource if
         parent_res_id is provided.
@@ -105,14 +114,18 @@ def retrieve_resource(self, res_type: APIResourceTypes, res_id: str, parent_res_
         :param parent_res_id:
         :param from_collection:
         :param collection_id:
+        :param url_endpoint: If given, will override the default URL construction. Should contain the endpoint past the API root.
         :return:
         """
-        url = self.resource_url_resolver(res_type, res_id, parent_res_id, from_collection)
+        if url_endpoint is None:
+            url = self.resource_url_resolver(res_type, None, parent_res_id, from_collection)
+        else:
+            url = f'{self.server_url}/{self.api_root}/{url_endpoint}'
         api_request = ConnectedSystemAPIRequest(url=url, request_method='GET', auth=self.get_helper_auth())
         return api_request.make_request()
 
     def update_resource(self, res_type: APIResourceTypes, res_id: str, json_data: any, parent_res_id: str = None,
-                        from_collection: bool = False):
+                        from_collection: bool = False, url_endpoint: str = None):
         """
         Updates a resource of the given type by its id, if necessary, will attempt to update a sub-resource if
         parent_res_id is provided.
@@ -121,31 +134,48 @@ def update_resource(self, res_type: APIResourceTypes, res_id: str, json_data: an
         :param json_data:
         :param parent_res_id:
         :param from_collection:
+        :param url_endpoint: If given, will override the default URL construction. Should contain the endpoint past the API root.
         :return:
         """
-        url = self.resource_url_resolver(res_type, None, parent_res_id, from_collection)
+        if url_endpoint is None:
+            url = self.resource_url_resolver(res_type, None, parent_res_id, from_collection)
+        else:
+            url = f'{self.server_url}/{self.api_root}/{url_endpoint}'
         api_request = ConnectedSystemAPIRequest(url=url, request_method='PUT', auth=self.get_helper_auth(),
                                                 body=json_data)
         return api_request.make_request()
 
     def delete_resource(self, res_type: APIResourceTypes, res_id: str, parent_res_id: str = None,
-                        from_collection: bool = False):
+                        from_collection: bool = False, url_endpoint: str = None):
         """
         Deletes a resource of the given type by its id, if necessary, will attempt to delete a sub-resource if
         parent_res_id is provided.
         :param res_type:
         :param res_id:
         :param parent_res_id:
         :param from_collection:
+        :param url_endpoint: If given, will override the default URL construction. Should contain the endpoint past the API root.
         :return:
         """
-        url = self.resource_url_resolver(res_type, res_id, parent_res_id, from_collection)
+        if url_endpoint is None:
+            url = self.resource_url_resolver(res_type, None, parent_res_id, from_collection)
+        else:
+            url = f'{self.server_url}/{self.api_root}/{url_endpoint}'
         api_request = ConnectedSystemAPIRequest(url=url, request_method='DELETE', auth=self.get_helper_auth())
         return api_request.make_request()
 
     # Helpers
     def resource_url_resolver(self, res_type: APIResourceTypes, res_id: str = None, parent_res_id: str = None,
                               from_collection: bool = False):
+        """
+        Helper to generate a URL endpoint for a given resource type and id by matching the resource type to an
+        appropriate parent endpoint and inserting the resource ids as necessary.
+        :param res_type:
+        :param res_id:
+        :param parent_res_id:
+        :param from_collection:
+        :return:
+        """
         if res_type is None:
             raise ValueError('Resource type must contain a valid APIResourceType')
         if res_type is APIResourceTypes.COLLECTION and from_collection:
@@ -160,8 +190,16 @@ def resource_url_resolver(self, res_type: APIResourceTypes, res_id: str = None,
         return self.construct_url(parent_type, res_id, res_type, parent_res_id)
 
     def construct_url(self, parent_type, res_id, res_type, parent_res_id):
+        """
+        Constructs an API endpoint url from the given parameters
+        :param parent_type:
+        :param res_id:
+        :param res_type:
+        :param parent_res_id:
+        :return:
+        """
         # TODO: Test for less common cases to ensure that the URL is being constructed correctly
-        base_url = f'{self.server_url}/{self.api_endpoint}'
+        base_url = f'{self.server_url}/{self.api_root}'
         resource_endpoint = resource_type_to_endpoint(res_type, parent_type)
         url = f'{base_url}/{resource_endpoint}'
 
@@ -178,3 +216,32 @@ def get_helper_auth(self):
         if self.user_auth:
             return self.username, self.password
         return None
+
+
+@dataclass(kw_only=True)
+class ResponseParserHelper:
+    default_object_reps: DefaultObjectRepresentations
+
+
+class DefaultObjectRepresentations(BaseModel):
+    """
+    Intended to be used as a way to determine which formats should be used when serializing and deserializing objects.
+    Should work in tandem with planned Serializer/Deserializer classes.
+    """
+    # Part 1
+    collections: str = Field(EncodingSchema.JSON.value)
+    deployments: str = Field(EncodingSchema. GEO_JSON.value)
+    procedures: str = Field(EncodingSchema.GEO_JSON.value)
+    properties: str = Field(EncodingSchema.SML_JSON.value)
+    sampling_features: str = Field(EncodingSchema.GEO_JSON.value)
+    systems: str = Field(EncodingSchema.GEO_JSON.value)
+    # Part 2
+    datastreams: str = Field(EncodingSchema.JSON.value)
+    observations: str = Field(EncodingSchema.JSON.value)
+    control_channels: str = Field(EncodingSchema.JSON.value)
+    commands: str = Field(EncodingSchema.JSON.value)
+    system_events: str = Field(EncodingSchema.OM_JSON.value)
+    system_history: str = Field(EncodingSchema.GEO_JSON.value)
+    # TODO: validate schemas for each resource to amke sure they are allowed per the spec
+
+