Moved to cleaner log translation organization

Author: neomatamune

cosmotech/coal/aws/s3.py

@@ -102,7 +102,7 @@ def upload_file(
         file_prefix: Prefix to add to the file name in the bucket
     """
     uploaded_file_name = file_prefix + file_path.name
-    LOGGER.info(T("coal.logs.data_transfer.file_sent").format(file_path=file_path, uploaded_name=uploaded_file_name))
+    LOGGER.info(T("coal.common.data_transfer.file_sent").format(file_path=file_path, uploaded_name=uploaded_file_name))
     s3_resource.Bucket(bucket_name).upload_file(str(file_path), uploaded_file_name)
 
 
@@ -125,8 +125,8 @@ def upload_folder(
     """
     source_path = pathlib.Path(source_folder)
     if not source_path.exists():
-        LOGGER.error(T("coal.errors.file_system.file_not_found").format(source_folder=source_folder))
-        raise FileNotFoundError(T("coal.errors.file_system.file_not_found").format(source_folder=source_folder))
+        LOGGER.error(T("coal.common.file_operations.not_found").format(source_folder=source_folder))
+        raise FileNotFoundError(T("coal.common.file_operations.not_found").format(source_folder=source_folder))
 
     if source_path.is_dir():
         _source_name = str(source_path)
@@ -135,7 +135,7 @@ def upload_folder(
                 _file_name = str(_file_path).removeprefix(_source_name).removeprefix("/")
                 uploaded_file_name = file_prefix + _file_name
                 LOGGER.info(
-                    T("coal.logs.data_transfer.file_sent").format(
+                    T("coal.common.data_transfer.file_sent").format(
                         file_path=_file_path, uploaded_name=uploaded_file_name
                     )
                 )
@@ -176,7 +176,7 @@ def download_files(
                 target_file = target_file.removeprefix(file_prefix)
             output_file = f"{target_folder}/{target_file}"
             pathlib.Path(output_file).parent.mkdir(parents=True, exist_ok=True)
-            LOGGER.info(T("coal.logs.storage.downloading").format(path=path_name, output=output_file))
+            LOGGER.info(T("coal.services.azure_storage.downloading").format(path=path_name, output=output_file))
             bucket.download_file(_file.key, output_file)
 
 
@@ -202,7 +202,7 @@ def upload_data_stream(
     size = len(data_stream.read())
     data_stream.seek(0)
 
-    LOGGER.info(T("coal.logs.data_transfer.sending_data").format(size=size))
+    LOGGER.info(T("coal.common.data_transfer.sending_data").format(size=size))
     s3_client.upload_fileobj(data_stream, bucket_name, uploaded_file_name)
 
 
@@ -228,8 +228,8 @@ def delete_objects(
 
     boto_objects = [{"Key": _file.key} for _file in bucket_files if _file.key != file_prefix]
     if boto_objects:
-        LOGGER.info(T("coal.logs.storage.deleting_objects").format(objects=boto_objects))
+        LOGGER.info(T("coal.services.azure_storage.deleting_objects").format(objects=boto_objects))
         boto_delete_request = {"Objects": boto_objects}
         bucket.delete_objects(Delete=boto_delete_request)
     else:
-        LOGGER.info(T("coal.logs.storage.no_objects"))
+        LOGGER.info(T("coal.services.azure_storage.no_objects"))

cosmotech/coal/azure/adx/auth.py

@@ -33,7 +33,7 @@ def create_kusto_client(
     Returns:
         KustoClient: A client for querying ADX
     """
-    LOGGER.debug(T("coal.logs.adx.creating_kusto_client").format(cluster_url=cluster_url))
+    LOGGER.debug(T("coal.services.adx.creating_kusto_client").format(cluster_url=cluster_url))
 
     try:
         az_client_id = client_id or os.environ["AZURE_CLIENT_ID"]
@@ -43,9 +43,9 @@ def create_kusto_client(
         kcsb = KustoConnectionStringBuilder.with_aad_application_key_authentication(
             cluster_url, az_client_id, az_client_secret, az_tenant_id
         )
-        LOGGER.debug(T("coal.logs.adx.using_app_auth"))
+        LOGGER.debug(T("coal.services.adx.using_app_auth"))
     except KeyError:
-        LOGGER.debug(T("coal.logs.adx.using_cli_auth"))
+        LOGGER.debug(T("coal.services.adx.using_cli_auth"))
         kcsb = KustoConnectionStringBuilder.with_az_cli_authentication(cluster_url)
 
     return KustoClient(kcsb)
@@ -69,7 +69,7 @@ def create_ingest_client(
     Returns:
         QueuedIngestClient: A client for ingesting data to ADX
     """
-    LOGGER.debug(T("coal.logs.adx.creating_ingest_client").format(ingest_url=ingest_url))
+    LOGGER.debug(T("coal.services.adx.creating_ingest_client").format(ingest_url=ingest_url))
 
     try:
         az_client_id = client_id or os.environ["AZURE_CLIENT_ID"]
@@ -79,9 +79,9 @@ def create_ingest_client(
         kcsb = KustoConnectionStringBuilder.with_aad_application_key_authentication(
             ingest_url, az_client_id, az_client_secret, az_tenant_id
         )
-        LOGGER.debug(T("coal.logs.adx.using_app_auth"))
+        LOGGER.debug(T("coal.services.adx.using_app_auth"))
     except KeyError:
-        LOGGER.debug(T("coal.logs.adx.using_cli_auth"))
+        LOGGER.debug(T("coal.services.adx.using_cli_auth"))
         kcsb = KustoConnectionStringBuilder.with_az_cli_authentication(ingest_url)
 
     return QueuedIngestClient(kcsb)
@@ -98,7 +98,7 @@ def initialize_clients(adx_uri: str, adx_ingest_uri: str) -> Tuple[KustoClient,
     Returns:
         tuple: (kusto_client, ingest_client)
     """
-    LOGGER.debug(T("coal.logs.adx.auth.initializing_clients"))
+    LOGGER.debug(T("coal.services.adx.initializing_clients"))
     kusto_client = create_kusto_client(adx_uri)
     ingest_client = create_ingest_client(adx_ingest_uri)
     return kusto_client, ingest_client
@@ -115,7 +115,9 @@ def get_cluster_urls(cluster_name: str, cluster_region: str) -> Tuple[str, str]:
     Returns:
         tuple: (cluster_url, ingest_url)
     """
-    LOGGER.debug(T("coal.logs.adx.generating_urls").format(cluster_name=cluster_name, cluster_region=cluster_region))
+    LOGGER.debug(
+        T("coal.services.adx.generating_urls").format(cluster_name=cluster_name, cluster_region=cluster_region)
+    )
 
     cluster_url = f"https://{cluster_name}.{cluster_region}.kusto.windows.net"
     ingest_url = f"https://ingest-{cluster_name}.{cluster_region}.kusto.windows.net"

cosmotech/coal/azure/adx/ingestion.py

@@ -64,7 +64,7 @@ def ingest_dataframe(
     Returns:
         The ingestion result with source_id for status tracking
     """
-    LOGGER.debug(T("coal.logs.adx.ingesting_dataframe").format(table_name=table_name, rows=len(dataframe)))
+    LOGGER.debug(T("coal.services.adx.ingesting_dataframe").format(table_name=table_name, rows=len(dataframe)))
 
     drop_by_tags = [drop_by_tag] if (drop_by_tag is not None) else None
 
@@ -83,7 +83,7 @@ def ingest_dataframe(
     _ingest_status[source_id] = IngestionStatus.QUEUED
     _ingest_times[source_id] = time.time()
 
-    LOGGER.debug(T("coal.logs.adx.ingestion_queued").format(source_id=source_id))
+    LOGGER.debug(T("coal.services.adx.ingestion_queued").format(source_id=source_id))
 
     return ingestion_result
 
@@ -112,10 +112,10 @@ def send_to_adx(
     Returns:
         The ingestion result with source_id for status tracking
     """
-    LOGGER.debug(T("coal.logs.adx.sending_to_adx").format(table_name=table_name, items=len(dict_list)))
+    LOGGER.debug(T("coal.services.adx.sending_to_adx").format(table_name=table_name, items=len(dict_list)))
 
     if not dict_list:
-        LOGGER.warning(T("coal.logs.adx.empty_dict_list"))
+        LOGGER.warning(T("coal.services.adx.empty_dict_list"))
         return None
 
     if not ignore_table_creation:
@@ -125,7 +125,7 @@ def send_to_adx(
 
         # Then try to create the table
         if not create_table(query_client, database, table_name, types):
-            LOGGER.error(T("coal.logs.adx.table_creation_failed").format(table_name=table_name))
+            LOGGER.error(T("coal.services.adx.table_creation_failed").format(table_name=table_name))
             return False
 
     # Create a dataframe with the data to write and send them to ADX
@@ -169,7 +169,7 @@ def check_ingestion_status(
     if not remaining_ids:
         return
 
-    LOGGER.debug(T("coal.logs.adx.checking_status").format(count=len(remaining_ids)))
+    LOGGER.debug(T("coal.services.adx.checking_status").format(count=len(remaining_ids)))
 
     # Get status queues
     qs = KustoIngestStatusQueues(client)
@@ -183,7 +183,7 @@ def get_messages(queues):
     successes = get_messages(qs.success._get_queues())
     failures = get_messages(qs.failure._get_queues())
 
-    LOGGER.debug(T("coal.logs.adx.status_messages").format(success=len(successes), failure=len(failures)))
+    LOGGER.debug(T("coal.services.adx.status_messages").format(success=len(successes), failure=len(failures)))
 
     queued_ids = list(remaining_ids)
     # Process success and failure messages
@@ -199,7 +199,7 @@ def get_messages(queues):
                 if dm.IngestionSourceId == str(source_id):
                     _ingest_status[source_id] = status
 
-                    log_function(T("coal.logs.adx.status_found").format(source_id=source_id, status=status.value))
+                    log_function(T("coal.services.adx.status_found").format(source_id=source_id, status=status.value))
 
                     _q.delete_message(_m)
                     remaining_ids.remove(source_id)
@@ -213,7 +213,7 @@ def get_messages(queues):
     for source_id in remaining_ids:
         if time.time() - _ingest_times[source_id] > actual_timeout:
             _ingest_status[source_id] = IngestionStatus.TIMEOUT
-            LOGGER.warning(T("coal.logs.adx.ingestion_timeout").format(source_id=source_id))
+            LOGGER.warning(T("coal.services.adx.ingestion_timeout").format(source_id=source_id))
 
     # Yield results for remaining IDs
     for source_id in queued_ids:
@@ -237,7 +237,7 @@ def monitor_ingestion(
     has_failures = False
     source_ids_copy = source_ids.copy()
 
-    LOGGER.info(T("coal.logs.adx.waiting_ingestion"))
+    LOGGER.info(T("coal.services.adx.waiting_ingestion"))
 
     with tqdm.tqdm(desc="Ingestion status", total=len(source_ids_copy)) as pbar:
         while any(
@@ -252,7 +252,7 @@ def monitor_ingestion(
             for ingestion_id, ingestion_status in results:
                 if ingestion_status == IngestionStatus.FAILURE:
                     LOGGER.error(
-                        T("coal.logs.adx.ingestion_failed").format(
+                        T("coal.services.adx.ingestion_failed").format(
                             ingestion_id=ingestion_id, table=table_ingestion_id_mapping.get(ingestion_id)
                         )
                     )
@@ -273,14 +273,14 @@ def monitor_ingestion(
             for ingestion_id, ingestion_status in results:
                 if ingestion_status == IngestionStatus.FAILURE:
                     LOGGER.error(
-                        T("coal.logs.adx.ingestion_failed").format(
+                        T("coal.services.adx.ingestion_failed").format(
                             ingestion_id=ingestion_id, table=table_ingestion_id_mapping.get(ingestion_id)
                         )
                     )
                     has_failures = True
         pbar.update(len(source_ids_copy))
 
-    LOGGER.info(T("coal.logs.adx.ingestion_completed"))
+    LOGGER.info(T("coal.services.adx.ingestion_completed"))
     return has_failures
 
 
@@ -298,7 +298,7 @@ def handle_failures(kusto_client: KustoClient, database: str, operation_tag: str
         bool: True if the process should abort, False otherwise
     """
     if has_failures:
-        LOGGER.warning(T("coal.logs.adx.failures_detected").format(operation_tag=operation_tag))
+        LOGGER.warning(T("coal.services.adx.failures_detected").format(operation_tag=operation_tag))
         _drop_by_tag(kusto_client, database, operation_tag)
         return True
     return False
@@ -314,10 +314,10 @@ def clear_ingestion_status_queues(client: QueuedIngestClient, confirmation: bool
         confirmation: Must be True to proceed with clearing
     """
     if not confirmation:
-        LOGGER.warning(T("coal.logs.adx.clear_queues_no_confirmation"))
+        LOGGER.warning(T("coal.services.adx.clear_queues_no_confirmation"))
         return
 
-    LOGGER.warning(T("coal.logs.adx.clearing_queues"))
+    LOGGER.warning(T("coal.services.adx.clearing_queues"))
     qs = KustoIngestStatusQueues(client)
 
     while not qs.success.is_empty():
@@ -326,4 +326,4 @@ def clear_ingestion_status_queues(client: QueuedIngestClient, confirmation: bool
     while not qs.failure.is_empty():
         qs.failure.pop(32)
 
-    LOGGER.info(T("coal.logs.adx.queues_cleared"))
+    LOGGER.info(T("coal.services.adx.queues_cleared"))

cosmotech/coal/azure/adx/query.py

@@ -24,11 +24,13 @@ def run_query(client: KustoClient, database: str, query: str) -> KustoResponseDa
     Returns:
         KustoResponseDataSet: The results of the query
     """
-    LOGGER.debug(T("coal.logs.adx.running_query").format(database=database, query=query))
+    LOGGER.debug(T("coal.services.adx.running_query").format(database=database, query=query))
 
     result = client.execute(database, query)
     LOGGER.debug(
-        T("coal.logs.adx.query_complete").format(rows=len(result.primary_results[0]) if result.primary_results else 0)
+        T("coal.services.adx.query_complete").format(
+            rows=len(result.primary_results[0]) if result.primary_results else 0
+        )
     )
 
     return result
@@ -46,9 +48,9 @@ def run_command_query(client: KustoClient, database: str, query: str) -> KustoRe
     Returns:
         KustoResponseDataSet: The results of the query
     """
-    LOGGER.debug(T("coal.logs.adx.running_command").format(database=database, query=query))
+    LOGGER.debug(T("coal.services.adx.running_command").format(database=database, query=query))
 
     result = client.execute_mgmt(database, query)
-    LOGGER.debug(T("coal.logs.adx.command_complete"))
+    LOGGER.debug(T("coal.services.adx.command_complete"))
 
     return result

cosmotech/coal/azure/adx/runner.py

@@ -57,7 +57,7 @@ def prepare_csv_content(folder_path: str) -> Dict[str, Dict[str, Any]]:
         cols = {k.strip(): "string" for k in headers}
         csv_datas = {"filename": _file.name.removesuffix(".csv"), "headers": cols}
         content[str(_file)] = csv_datas
-    LOGGER.debug(T("coal.logs.adx.runner.content_debug").format(content=content))
+    LOGGER.debug(T("coal.services.adx.content_debug").format(content=content))
 
     return content
 
@@ -110,7 +110,7 @@ def insert_csv_files(
         fields = file_info.get("headers")
         with open(file_path) as _f:
             file_size = sum(map(len, _f.readlines()))
-            LOGGER.debug(T("coal.logs.data_transfer.sending_data").format(size=file_size))
+            LOGGER.debug(T("coal.common.data_transfer.sending_data").format(size=file_size))
         fd = FileDescriptor(file_path, file_size)
         ord = 0
         mappings = list()