Improve documentation from CLI

Author: SprinTech

pipelines/notebooks/test_plusierus_annees.ipynb

@@ -21,14 +21,22 @@ def cli():
 def list():
     """List all available tasks."""
     tasks_dir = os.path.join(os.path.dirname(__file__), "tasks")
-    click.echo("Available tasks:")
-    for filename in os.listdir(tasks_dir):
+
+    for filename in sorted(os.listdir(tasks_dir)):
         if filename.endswith(".py") and not filename.startswith("_"):
             module_name = filename[:-3]
             module = importlib.import_module(f"tasks.{module_name}")
-            description = module.__doc__ or "No description"
-            description = description.strip().split("\n")[0]
-            click.echo(f"- {module_name}: {description}")
+
+            doc = module.__doc__ or "No description"
+            doc_lines = doc.strip().split("\n")
+            while doc_lines and not doc_lines[0].strip():
+                doc_lines.pop(0)
+            while doc_lines and not doc_lines[-1].strip():
+                doc_lines.pop()
+
+            click.echo(f"\n{module_name}:")
+            for line in doc_lines:
+                click.echo(f"    {line}")
 
 
 @cli.command()
@@ -38,27 +46,29 @@ def list():
     type=click.Choice(["all", "last", "custom"]),
     default="all",
     help="Type of refresh to perform",
+    hidden=True,
 )
 @click.option(
     "--custom-years",
     type=str,
     help="Comma-separated list of years to process (for custom refresh type)",
+    hidden=True,
 )
-def run(task_name, refresh_type, custom_years):
+def run(task_name, refresh_type, custom_years, drop_tables):
     """Run a specified task."""
     try:
         module = importlib.import_module(f"tasks.{task_name}")
         task_func = getattr(module, "execute")
         logging.info(f"Starting task {task_name}...")
-        
+
         # Parse custom years if provided
         custom_years_list = None
         if custom_years:
             custom_years_list = [year.strip() for year in custom_years.split(",")]
-        
+
         # Call the task function with parameters
         task_func(refresh_type=refresh_type, custom_years=custom_years_list)
-        
+
         logging.info(f"Task {task_name} completed.")
     except (ModuleNotFoundError, AttributeError):
         logging.error(f"Task {task_name} not found.")

pipelines/run.py

@@ -1,17 +1,20 @@
 """
-Consolidate data into the database.
+Consolidate data into the database.\n
+Args:
+    - refresh-type (str): Type of refresh to perform ("all", "last", or "custom")
+    - custom-years (list[str]): List of years to process when refresh_type is "custom"
 """
 
 import logging
 import os
-from typing import Dict, List, Literal
+from typing import List, Literal
 from zipfile import ZipFile
 
 import duckdb
 import requests
 
 from ._common import CACHE_FOLDER, DUCKDB_FILE, clear_cache
-from ._config_edc import get_edc_config, create_edc_yearly_filename
+from ._config_edc import create_edc_yearly_filename, get_edc_config
 
 logger = logging.getLogger(__name__)
 edc_config = get_edc_config()
@@ -33,7 +36,6 @@ def check_table_existence(conn: duckdb.DuckDBPyConnection, table_name: str) -> b
     return list(conn.fetchone())[0] == 1
 
 
-
 def download_extract_insert_yearly_edc_data(year: str):
     """
     Downloads from www.data.gouv.fr the EDC (Eau distribuée par commune) dataset for one year,
@@ -134,7 +136,9 @@ def process_edc_datasets(
                     f"Invalid years provided: {sorted(invalid_years)}. Years must be among: {available_years}"
                 )
             # Filtering and sorting of valid years
-            years_to_update = sorted(list(set(custom_years).intersection(available_years)))
+            years_to_update = sorted(
+                list(set(custom_years).intersection(available_years))
+            )
         else:
             raise ValueError(
                 """ custom_years parameter needs to be specified if refresh_type="custom" """
@@ -153,10 +157,11 @@ def process_edc_datasets(
     clear_cache(recreate_folder=False)
     return True
 
+
 def execute(refresh_type: str = "all", custom_years: List[str] = None):
     """
     Execute the EDC dataset processing with specified parameters.
-    
+
     :param refresh_type: Type of refresh to perform ("all", "last", or "custom")
     :param custom_years: List of years to process when refresh_type is "custom"
     """