authentication commands setting

The adjustment is made to add the authentication commands such as user and password, internally accessing the generated token and making the request to the server

README.md

@@ -64,12 +64,27 @@ Para crear una colección siga los siguientes pasos:
 
 # Instrucciones de Uso
 
+## Autenticación Obligatoria
+
+Todos los comandos requieren autenticación con las opciones `-u` y `-p` para proporcionar el nombre de usuario y la contraseña.
+
+### Parámetros Globales:
+- `-u, --username` (obligatorio): Nombre de usuario para autenticación.
+- `-p, --password` (obligatorio): Contraseña para autenticación.
+
+Ejemplo de autenticación:
+
+
+python src/main.py -u admin1 -p admin [subcomando] [opciones]
+
+---
+
 ## Cargar una Colección
 
 Para cargar una colección de capas, ejecuta el siguiente comando:
 
 ```
-python src/main.py create -f folder_name [-c collection_name] [-o]
+python src/main.py -u username -p password create -f folder_name [-c collection_name]
 ```
 
 ### Parámetros:
@@ -80,24 +95,23 @@ python src/main.py create -f folder_name [-c collection_name] [-o]
 
 * Especificando un nombre de colección
 ```
-python src/main.py create -f my_folder -c MyCollection
+python src/main.py -u username -p password create -f my_folder -c MyCollection
 
 o
 
-python src/main.py create --folder my_folder --collection MyCollection
-
+python src/main.py -u username -p password create --folder my_folder --collection MyCollection
 ```
 
 Este comando creará la colección `MyCollection` a partir de los archivos en el directorio `input/my_folder`
 
 * Usando el id del archivo collection.json:
 
 ```
-python src/main.py create -f my_folder
+python src/main.py -u username -p password create -f my_folder
 
 o
 
-python src/main.py create --folder my_folder
+python src/main.py -u username -p password create --folder my_folder
 
 ```
 ---
@@ -107,7 +121,7 @@ python src/main.py create --folder my_folder
 Si solo deseas validar una colección sin cargarla, puedes ejecutar:
 
 ```
-python src/main.py validate -f folder_name [-c collection_name]
+python src/main.py -u username -p password validate -f folder_name [-c collection_name]
 ```
 
 ### Parámetros:
@@ -117,12 +131,11 @@ python src/main.py validate -f folder_name [-c collection_name]
 ### Ejemplo:
 
 ```
-python src/main.py validate -f my_folder
+python src/main.py -u username -p password validate -f my_folder
 
 o
 
-python src/main.py validate --folder my_folder
-
+python src/main.py -u username -p password validate --folder my_folder
 ```
 
 Este comando validará los archivos de la colección en el directorio `input/my_folder` sin cargarlos.
@@ -134,7 +147,7 @@ Este comando validará los archivos de la colección en el directorio `input/my_
 Para eliminar una colección de STAC y de Azure, ejecuta el siguiente comando:
 
 ```
-python src/main.py remove --collection collection_name
+python src/main.py -u username -p password remove --collection collection_name
 ```
 
 ### Parámetros:
@@ -144,12 +157,11 @@ python src/main.py remove --collection collection_name
 ### Ejemplo:
 
 ```
-python src/main.py remove -c my_collection
+python src/main.py -u username -p password remove -c my_collection
 
 o
 
-python src/main.py remove --collection my_collection
-
+python src/main.py -u username -p password remove --collection my_collection
 ```
 
 Este comando eliminará la colección `my_collection` del sistema.
@@ -168,7 +180,6 @@ Para hacer formateo de estilos automático se utiliza el paquete black. Al ejecu
 
 ```
 black src
-
 ```
 
 ## Documentación

src/collection.py

@@ -11,7 +11,7 @@
 
 class Collection:
 
-    def __init__(self):
+    def __init__(self,token: str):
         self.items = []
         self.dates = []
         self.longs = []
@@ -20,6 +20,8 @@ def __init__(self):
         self.stac_items = []
         self.stac_url = get_settings().stac_url
         self.storage = storage.Storage()
+        self.token = token
+        self.headers = {"Authorization": f"Bearer {self.token}"}
 
     def load_items(self, folder, raw_items):
         """
@@ -126,7 +128,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)
+        exist = stac_rest.check_resource(url,headers=self.headers)
         if exist:
             collection_exist = True
             if not overwritten:
@@ -156,13 +158,13 @@ def remove_collection(self, collection_name=None):
         logger.info(f"Attempting to remove collection {collection_id}")
 
         try:
-            items_collection = stac_rest.get(f"{collection_url}/items").json()
-            for item in items_collection["features"]:
+            items_response = stac_rest.get(f"{collection_url}/items", headers=self.headers)
+            items = items_response.json().get("features", [])
+
+            for item in items:
                 for asset_key, asset_value in item["assets"].items():
                     url = asset_value["href"]
-                    logger.info(
-                        f"Deleting file: {url} from Azure Blob Storage"
-                    )
+                    logger.info(f"Deleting file {url} from Azure Blob Storage")
                     self.storage.remove_file(url)
 
             stac_rest.delete(collection_url)
@@ -184,6 +186,7 @@ def upload_collection(self):
             stac_rest.post_or_put(
                 parse.urljoin(self.stac_url, "/collections"),
                 self.stac_collection.to_dict(),
+                headers=self.headers,
             )
             logger.info(
                 f"Collection {self.stac_collection.id} uploaded successfully"
@@ -197,6 +200,7 @@ def upload_collection(self):
                         f"/collections/{self.stac_collection.id}/items",
                     ),
                     item_dict,
+                    headers=self.headers
                 )
                 logger.info(
                     f"Item upload response: {item_response.status_code}"

src/config.py

@@ -6,6 +6,7 @@ class Settings(BaseSettings):
     stac_url: str = "http://localhost:8082"
     abs_string: str = ""
     abs_container: str = "cog-test"
+    auth_url: str
 
     model_config = SettingsConfigDict(env_file=".env")
 

src/main.py

@@ -1,11 +1,16 @@
 from argparse import ArgumentParser
 
 from utils.logging_config import logger
+from pypgstac.db import settings
 from utils import spec
+from utils.auth import authenticate
 from collection import Collection
 from json import load
 from sys import exit as sysexit
 from os import getcwd
+from config import get_settings
+
+settings = get_settings()
 
 
 def create_collection_local(collection, input_folder, collection_name):
@@ -20,6 +25,7 @@ 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.")
 
@@ -28,105 +34,64 @@ def create_collection_local(collection, input_folder, collection_name):
 
 
 def main():
-    """
-    Read arguments and start data validation.
-    """
-
-    parser = ArgumentParser()
-    sub_parsers = parser.add_subparsers(dest="command")
-
-    create_parser = sub_parsers.add_parser(
-        "create", help="Create a new collection"
-    )
-    create_parser.add_argument(
-        "-f",
-        "--folder",
-        dest="folder",
-        help="Collection folder",
-        required=True,
-    )
-    create_parser.add_argument(
-        "-c",
-        "--collection",
-        dest="collection",
-        help="Collection name",
-        required=False,
-    )
-    create_parser.add_argument(
-        "-o",
-        "--overwrite",
-        dest="overwrite",
-        action="store_true",
-        help="Overwrite existing collection",
-        required=False,
-    )
-
-    validate_parser = sub_parsers.add_parser(
-        "validate", help="Validate collection specification"
-    )
-    validate_parser.add_argument(
-        "-f",
-        "--folder",
-        dest="folder",
-        help="Collection folder",
-        required=True,
-    )
-    validate_parser.add_argument(
-        "-c",
-        "--collection",
-        dest="collection",
-        help="Collection name",
-        required=False,
-    )
-
-    remove_parser = sub_parsers.add_parser(
-        "remove", help="Remove indicated collection"
-    )
-    remove_parser.add_argument(
-        "-c",
-        "--collection",
-        dest="collection",
-        help="Collection name",
-        required=True,
-    )
-
-    args = parser.parse_args()
-
-    if args.command == "create":
-        collection = Collection()
-        input_folder = f"input/{args.folder}"
-        create_collection_local(collection, input_folder, args.collection)
-
-        if collection.check_collection(args.overwrite):
-            collection.remove_collection()
-            sysexit("Previous collection removed successfully.")
-
-        output_dir = f"{getcwd()}/output/{args.folder}"
-        collection.convert_layers(input_folder, output_dir)
-        logger.info("Layers converted successfully.")
-
-        collection.upload_layers(output_dir)
-        logger.info("Layers uploaded successfully.")
-
-        collection.upload_collection()
-        logger.info("Collection uploaded successfully.")
-
-        sysexit("Process completed successfully.")
-
-    elif args.command == "validate":
-        collection = Collection()
-        create_collection_local(
-            collection, f"input/{args.folder}", args.collection
-        )
-        sysexit("Validation successful.")
-
-    elif args.command == "remove":
-        collection = Collection()
-        collection.remove_collection(args.collection)
-        sysexit("Collection successfully removed.")
-
-    else:
-        sysexit("No command used. Type -h for help")
+
+    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")
+    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("-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")
+    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")
+
+    try:
+        args = parser.parse_args()
+
+        token = authenticate(args.username, args.password, settings.stac_url, settings.auth_url)
+
+        collection = Collection(token)
+
+        if args.command == "create":
+            input_folder = f"input/{args.folder}"
+            create_collection_local(collection, input_folder, args.collection)
+
+            if collection.check_collection(args.overwrite):
+                collection.remove_collection()
+                logger.info("Previous collection removed.")
+
+            output_dir = f"{getcwd()}/output/{args.folder}"
+            collection.convert_layers(input_folder, output_dir)
+            logger.info("Layers converted successfully.")
+
+            collection.upload_layers(output_dir)
+            collection.upload_collection()
+            logger.info("Collection uploaded successfully.")
+
+        elif args.command == "validate":
+            input_folder = f"input/{args.folder}"
+            create_collection_local(collection, input_folder, args.collection)
+            logger.info("Validation successful.")
+
+        elif args.command == "remove":
+            collection.remove_collection(args.collection)
+            logger.info("Collection removed successfully.")
+
+        else:
+            sysexit("No command used. Type -h for help")
+
+    except SystemExit as e:
+        logger.info("Error en los argumentos:", e)
 
 
 if __name__ == "__main__":

src/test.py

@@ -0,0 +1,16 @@
+# test.py
+from argparse import ArgumentParser
+
+
+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")
+
+    args = parser.parse_args()
+    print(f"Username: {args.username}")
+    print(f"Password: {args.password}")
+
+
+if __name__ == "__main__":
+    main()