diff --git a/geomagio/metadata/MetadataFactory.py b/geomagio/metadata/MetadataFactory.py index 92b449752f6739dab281b9280dfa6ffe86414972..9da666841e7bfc4773f7d1e060d101bf8c562e00 100644 --- a/geomagio/metadata/MetadataFactory.py +++ b/geomagio/metadata/MetadataFactory.py @@ -9,7 +9,7 @@ from pydantic import parse_obj_as from .Metadata import Metadata from .MetadataQuery import MetadataQuery - +# Set the API host and URL for the Geomagnetic Metadata service GEOMAG_API_HOST = os.getenv("GEOMAG_API_HOST", "geomag.usgs.gov") GEOMAG_API_URL = f"https://{GEOMAG_API_HOST}/ws/secure/metadata" if "127.0.0.1" in GEOMAG_API_URL: @@ -17,15 +17,36 @@ if "127.0.0.1" in GEOMAG_API_URL: class MetadataFactory(object): + """ + A factory class for interacting with metadata servers. + + Attributes: + url (str): The base URL for the metadata server. + token (str): The authorization token for accessing the API. + """ + def __init__( self, url: str = GEOMAG_API_URL, token: str = os.getenv("GITLAB_API_TOKEN"), ): + """ + Initializes the MetadataFactory with the specified URL and token. + + Args: + url (str): The base URL for the metadata server (default is GEOMAG_API_URL). + token (str): The authorization token (default is fetched from environment). + """ self.url = url self.token = token def _get_headers(self): + """ + Constructs the headers for the API request. + + Returns: + dict: A dictionary containing the authorization token and content type. + """ return ( {"Authorization": self.token, "content-type": "application/json"} if self.token @@ -33,6 +54,15 @@ class MetadataFactory(object): ) def get_metadata(self, query: MetadataQuery) -> List[Metadata]: + """ + Retrieves metadata based on the provided query. + + Args: + query (MetadataQuery): The query object containing search parameters. + + Returns: + List[Metadata]: A list of Metadata objects matching the query. + """ if query.id: metadata = [self.get_metadata_by_id(id=query.id)] else: @@ -43,7 +73,7 @@ class MetadataFactory(object): ) metadata = parse_obj_as(List[Metadata], response.json()) - # if metadata dict() provided in query, only include Metadata objects with matching key-value pairs + # Filter metadata based on additional criteria if provided if query.metadata is not None: query_metadata_items = query.metadata.items() metadata_list = metadata @@ -60,6 +90,15 @@ class MetadataFactory(object): return metadata def get_metadata_by_id(self, id: int) -> Metadata: + """ + Retrieves metadata for a specific ID. + + Args: + id (int): The ID of the metadata to retrieve. + + Returns: + Metadata: The Metadata object corresponding to the given ID. + """ response = requests.get( url=f"{self.url}/{id}", headers=self._get_headers(), @@ -67,6 +106,15 @@ class MetadataFactory(object): return Metadata(**response.json()) def create_metadata(self, metadata: Metadata) -> Metadata: + """ + Creates new metadata objects in the database. + + Args: + metadata (Metadata): The Metadata object to create. + + Returns: + Metadata: The created Metadata object. + """ response = requests.post( url=self.url, data=metadata.json(), @@ -75,6 +123,15 @@ class MetadataFactory(object): return Metadata(**response.json()) def update_metadata(self, metadata: Metadata) -> Metadata: + """ + Updates existing metadata in the system. + + Args: + metadata (Metadata): The Metadata object to update. + + Returns: + Metadata: The updated Metadata object. + """ response = requests.put( url=f"{self.url}/{metadata.id}", data=metadata.json(), @@ -84,17 +141,26 @@ class MetadataFactory(object): def parse_params(query: MetadataQuery) -> str: + """ + Parses the query object into a dictionary of parameters for the API request. + + Args: + query (MetadataQuery): The query object containing search parameters. + + Returns: + dict: A dictionary of parameters formatted for the API request. + """ query = query.dict(exclude_none=True) args = {} for key in query.keys(): element = query[key] - # convert times to strings + # Convert times to ISO format strings if isinstance(element, UTCDateTime): element = element.isoformat() - # serialize the metadata dictionary to a JSON string + # Serialize the metadata dictionary to a JSON string elif key == "metadata" and isinstance(element, dict): args[key] = json.dumps(element) - # get string value of metadata category + # Get string value of metadata category if key == "category": element = element.value args[key] = element