adjust format code

PEM-Humboldt · Nov 25, 2024 · 7b46050 · 7b46050
1 parent 977a263
commit 7b46050
Show file tree

Hide file tree

Showing 5 changed files with 56 additions and 42 deletions.
diff --git a/src/collection.py b/src/collection.py
@@ -11,7 +11,7 @@
 
 class Collection:
 
-    def __init__(self,token: str):
+    def __init__(self, token: str):
         self.items = []
         self.dates = []
         self.longs = []
@@ -97,9 +97,7 @@ def create_collection(self, collection_name, collection_data):
             extent=pystac.Extent(
                 spatial=spatial_extent, temporal=temporal_extent
             ),
-            extra_fields={
-                "metadata": collection_data["metadata"]
-            },
+            extra_fields={"metadata": collection_data["metadata"]},
         )
 
         self.stac_collection.validate()
@@ -128,7 +126,7 @@ def check_collection(self, overwritten):
         Check if the collection exists and if it's going to be overwritten.
         """
         url = f"{self.stac_url}/collections/{self.stac_collection.id}"
-        exist = stac_rest.check_resource(url,headers=self.headers)
+        exist = stac_rest.check_resource(url, headers=self.headers)
         if exist:
             collection_exist = True
             if not overwritten:
@@ -158,7 +156,9 @@ def remove_collection(self, collection_name=None):
         logger.info(f"Attempting to remove collection {collection_id}")
 
         try:
-            items_response = stac_rest.get(f"{collection_url}/items", headers=self.headers)
+            items_response = stac_rest.get(
+                f"{collection_url}/items", headers=self.headers
+            )
             items = items_response.json().get("features", [])
 
             for item in items:
@@ -200,7 +200,7 @@ def upload_collection(self):
                         f"/collections/{self.stac_collection.id}/items",
                     ),
                     item_dict,
-                    headers=self.headers
+                    headers=self.headers,
                 )
                 logger.info(
                     f"Item upload response: {item_response.status_code}"

diff --git a/src/main.py b/src/main.py
@@ -25,7 +25,6 @@ def create_collection_local(collection, input_folder, collection_name):
 
     collection.load_items(input_folder, raw_items)
 
-
     collection.create_collection(collection_name, data)
     logger.info("Collection created successfully.")
 
@@ -36,29 +35,42 @@ def create_collection_local(collection, input_folder, collection_name):
 def main():
 
     parser = ArgumentParser(description="STAC Collection Manager")
-    parser.add_argument("-u", "--username", required=True, help="Username for authentication")
-    parser.add_argument("-p", "--password", required=True, help="Password for authentication")
+    parser.add_argument(
+        "-u", "--username", required=True, help="Username for authentication"
+    )
+    parser.add_argument(
+        "-p", "--password", required=True, help="Password for authentication"
+    )
     subparsers = parser.add_subparsers(dest="command", help="Commands")
 
-    # Create Command
     create_parser = subparsers.add_parser("create", help="Create a collection")
-    create_parser.add_argument("-f", "--folder", required=True, help="Input folder")
+    create_parser.add_argument(
+        "-f", "--folder", required=True, help="Input folder"
+    )
     create_parser.add_argument("-c", "--collection", help="Collection name")
-    create_parser.add_argument("-o", "--overwrite", action="store_true", help="Overwrite")
-
-    # Validate Command
-    validate_parser = subparsers.add_parser("validate", help="Validate a collection")
-    validate_parser.add_argument("-f", "--folder", required=True, help="Input folder")
+    create_parser.add_argument(
+        "-o", "--overwrite", action="store_true", help="Overwrite"
+    )
+
+    validate_parser = subparsers.add_parser(
+        "validate", help="Validate a collection"
+    )
+    validate_parser.add_argument(
+        "-f", "--folder", required=True, help="Input folder"
+    )
     validate_parser.add_argument("-c", "--collection", help="Collection name")
 
-    # Remove Command
     remove_parser = subparsers.add_parser("remove", help="Remove a collection")
-    remove_parser.add_argument("-c", "--collection", required=True, help="Collection name")
+    remove_parser.add_argument(
+        "-c", "--collection", required=True, help="Collection name"
+    )
 
     try:
         args = parser.parse_args()
 
-        token = authenticate(args.username, args.password, settings.stac_url, settings.auth_url)
+        token = authenticate(
+            args.username, args.password, settings.stac_url, settings.auth_url
+        )
 
         collection = Collection(token)
 

diff --git a/src/test.py b/src/test.py
@@ -4,8 +4,12 @@
 
 def main():
     parser = ArgumentParser(description="STAC Collection Manager")
-    parser.add_argument("-u", "--username", required=True, help="Username for authentication")
-    parser.add_argument("-p", "--password", required=True, help="Password for authentication")
+    parser.add_argument(
+        "-u", "--username", required=True, help="Username for authentication"
+    )
+    parser.add_argument(
+        "-p", "--password", required=True, help="Password for authentication"
+    )
 
     args = parser.parse_args()
     print(f"Username: {args.username}")

diff --git a/src/utils/auth.py b/src/utils/auth.py
@@ -8,15 +8,10 @@ def authenticate(username, password, stac_url, auth_url):
     Authenticate against the authentication service and retrieve a token.
     """
     try:
-        auth_data = {
-            "username": username,
-            "password": password
-        }
+        auth_data = {"username": username, "password": password}
 
-        response = requests.post(
-            f"{stac_url}{auth_url}", data=auth_data
-        )
-        response.raise_for_status()  # Lanza un error si la respuesta tiene un código de error HTTP
+        response = requests.post(f"{stac_url}{auth_url}", data=auth_data)
+        response.raise_for_status()
 
         token = response.json().get("access_token")
 
@@ -26,14 +21,17 @@ def authenticate(username, password, stac_url, auth_url):
 
         return token
 
-
-
     except requests.HTTPError as http_err:
         error_detail = None
 
-        status_code = response.status_code if response is not None else "Unknown"
+        status_code = (
+            response.status_code if response is not None else "Unknown"
+        )
 
-        if response is not None and response.headers.get('Content-Type') == 'application/json':
+        if (
+            response is not None
+            and response.headers.get("Content-Type") == "application/json"
+        ):
 
             try:
                 error_detail = response.json().get("detail", str(http_err))
@@ -46,6 +44,5 @@ def authenticate(username, password, stac_url, auth_url):
 
         sysexit(f"Authentication error: {error_message}")
 
-
     except requests.RequestException as req_err:
         sysexit(f"Request error: {req_err}")
diff --git a/src/utils/stac_rest.py b/src/utils/stac_rest.py
@@ -1,15 +1,16 @@
 import requests
 
-def post_or_put(url: str, data: dict,headers: dict = None):
+
+def post_or_put(url: str, data: dict, headers: dict = None):
     """
     Post or put data to URL
     """
     try:
-        response = requests.post(url, json=data,headers=headers)
+        response = requests.post(url, json=data, headers=headers)
 
         # Si el POST falla con 409, intenta con PUT
         if response.status_code == 409:
-            response = requests.put(url, json=data,headers=headers)
+            response = requests.put(url, json=data, headers=headers)
 
         response.raise_for_status()
         return response
@@ -18,7 +19,7 @@ def post_or_put(url: str, data: dict,headers: dict = None):
         raise e
 
 
-def get(url: str,headers: dict = None):
+def get(url: str, headers: dict = None):
     """
     Get request
     """
@@ -28,11 +29,11 @@ def get(url: str,headers: dict = None):
     return response
 
 
-def check_resource(url: str,headers: dict = None):
+def check_resource(url: str, headers: dict = None):
     """
     Check if an URL for a resource exists. e.g. items, collections, catalogues
     """
-    response = requests.get(url,headers=headers)
+    response = requests.get(url, headers=headers)
     print(response)
     if response.status_code == 200:
         success = True
@@ -43,7 +44,7 @@ def check_resource(url: str,headers: dict = None):
     return success
 
 
-def delete(url,headers: dict = None):
+def delete(url, headers: dict = None):
     """
     Delete request
     """