fix(biomark): cant find duckdb file

Author: taheeraahmed

src/datasync/pit_registering_salmon.py

@@ -2,7 +2,6 @@
 
 """Biomark PIT registering salmon data synchronization."""
 
-import os
 from datetime import datetime, timedelta
 from pathlib import Path
 
@@ -361,106 +360,12 @@ def run(
         )
     )
 
-
-def create_stream_config(config: dict, location: str) -> dict:
-    """Create a stream configuration for a specific data type and location."""
-    return {
-        "primary_key": config["primary_key"],
-        "update_key": config["update_key"],
-        "object": (
-            f"tables/{config['table_name']}/location={location}/"
-            f"{{part_year}}/{{part_month}}/{{part_day}}"
-        ),
-        "mode": "incremental",
-        "target_options": {"format": "parquet"},
-        "sql": (
-            f"select * replace ("  # noqa: S608
-            f"{config['time_column']} at time zone 'UTC' as "
-            f"{config['time_column']}) "
-            f"from {config['table_name']} where "
-            f"{config['location']} = '{location}' "
-            f"and {{incremental_where_cond}}"
-        ),
-    }
-
-
-def check_table_has_data(db_path: str, table_name: str, dataset_name: str) -> bool:
-    """
-    Check if a table exists and has data in the DuckDB database.
-
-    Args:
-        db_path: Path to the DuckDB database file
-        table_name: Name of the table to check
-        dataset_name: Name of the schema/dataset
-
-    Returns:
-        bool: True if table exists and has at least one row, False otherwise
-    """
-    try:
-        conn = duckdb.connect(db_path, read_only=True)
-        try:
-            log.info(f"Checking table {table_name} existence in schema {dataset_name}")
-
-            check_table_exists = """
-                SELECT table_name
-                FROM information_schema.tables
-                WHERE table_schema = ? AND table_name = ?
-            """
-            result = conn.execute(
-                check_table_exists, [dataset_name, table_name]
-            ).fetchone()
-
-            if result is None:
-                log.info(f"Table {table_name} does not exist in schema {dataset_name}")
-                return False
-
-            qualified_table = f"{dataset_name}.{table_name}"
-            table_relation = conn.table(qualified_table)
-            count_result = table_relation.count("*").fetchone()
-
-            row_count = count_result[0] if count_result else 0
-
-            log.info(f"Table {table_name} has {row_count} rows")
-            return row_count > 0
-
-        finally:
-            conn.close()
-
-    except Exception as e:
-        log.error(f"Error checking table {table_name}: {e}")
-        return False
-
-
-@app.command()
-def replicate(
-    bucket: str = BIOMARK_BUCKET,
-    endpoint_url: str = BIOMARK_AWS_ENDPOINT,
-    access_key: str = BIOMARK_ACCESS_KEY,
-    secret_key: str = BIOMARK_SECRET_KEY,
-    duckdb_path: str = BIOMARK_DUCKDB_PATH,
-    region: str = BIOMARK_REGION,
-    dataset_name: str = BIOMARK_DATASET_NAME,
-    tags: bool = typer.Option(False, help="Add tags data to S3"),
-    readers: bool = typer.Option(False, help="Add readers voltage data to S3"),
-    environment: bool = typer.Option(False, help="Add environment data to S3"),
-):
-    """Upload data from .duckdb to S3 bucket."""
-    os.environ["NINAS3"] = f"{{type: s3, bucket: {bucket}, use_environment: true }}"
-    os.environ["AWS_ACCESS_KEY_ID"] = access_key
-    os.environ["AWS_SECRET_ACCESS_KEY"] = secret_key
-    os.environ["AWS_REGION"] = region
-    os.environ["AWS_ENDPOINT"] = endpoint_url
-    os.environ["DUCKDB"] = f"{{type: duckdb, instance: {duckdb_path}}}"
-
-    if not any([readers, tags, environment]):
+    if not Path(duckdb_path).exists():
         raise typer.BadParameter(
-            "Error: At least one data type must be selected "
-            "(--readers, --tags, or --environment)"
+            f"Error: DuckDB file not found at {Path(duckdb_path).resolve()}"
         )
 
-    if not Path(duckdb_path).exists():
-        raise typer.BadParameter(f"Error: DuckDB file not found at {duckdb_path}")
-
+    ## REPLICATE WITH SLING
     # only include configs for enabled data types that have data
     stream_configs = {}
     if readers and check_table_has_data(duckdb_path, "readers_voltage", dataset_name):
@@ -525,10 +430,87 @@ def replicate(
             "target_options": {"format": "parquet"},
         },
         streams=streams,
-        env={"SLING_STATE": "NINAS3/data/sling", "SLING_DIRECT_INSERT": "True"},
+        env={
+            "SLING_STATE": "NINAS3/data/sling",
+            "SLING_DIRECT_INSERT": "True",
+            "NINAS3": f"{{type: s3, bucket: {BIOMARK_BUCKET}, use_environment: true }}",
+            "AWS_ACCESS_KEY_ID": BIOMARK_ACCESS_KEY,
+            "AWS_SECRET_ACCESS_KEY": BIOMARK_SECRET_KEY,
+            "AWS_REGION": BIOMARK_REGION,
+            "AWS_ENDPOINT": BIOMARK_AWS_ENDPOINT,
+        },
         debug=True,
     ).run()
 
 
+def create_stream_config(config: dict, location: str) -> dict:
+    """Create a stream configuration for a specific data type and location."""
+    return {
+        "primary_key": config["primary_key"],
+        "update_key": config["update_key"],
+        "object": (
+            f"tables/{config['table_name']}/location={location}/"
+            f"{{part_year}}/{{part_month}}/{{part_day}}"
+        ),
+        "mode": "incremental",
+        "target_options": {"format": "parquet"},
+        "sql": (
+            f"select * replace ("  # noqa: S608
+            f"{config['time_column']} at time zone 'UTC' as "
+            f"{config['time_column']}) "
+            f"from {config['table_name']} where "
+            f"{config['location']} = '{location}' "
+            f"and {{incremental_where_cond}}"
+        ),
+    }
+
+
+def check_table_has_data(db_path: str, table_name: str, dataset_name: str) -> bool:
+    """
+    Check if a table exists and has data in the DuckDB database.
+
+    Args:
+        db_path: Path to the DuckDB database file
+        table_name: Name of the table to check
+        dataset_name: Name of the schema/dataset
+
+    Returns:
+        bool: True if table exists and has at least one row, False otherwise
+    """
+    try:
+        conn = duckdb.connect(db_path, read_only=True)
+        try:
+            log.info(f"Checking table {table_name} existence in schema {dataset_name}")
+
+            check_table_exists = """
+                SELECT table_name
+                FROM information_schema.tables
+                WHERE table_schema = ? AND table_name = ?
+            """
+            result = conn.execute(
+                check_table_exists, [dataset_name, table_name]
+            ).fetchone()
+
+            if result is None:
+                log.info(f"Table {table_name} does not exist in schema {dataset_name}")
+                return False
+
+            qualified_table = f"{dataset_name}.{table_name}"
+            table_relation = conn.table(qualified_table)
+            count_result = table_relation.count("*").fetchone()
+
+            row_count = count_result[0] if count_result else 0
+
+            log.info(f"Table {table_name} has {row_count} rows")
+            return row_count > 0
+
+        finally:
+            conn.close()
+
+    except Exception as e:
+        log.error(f"Error checking table {table_name}: {e}")
+        return False
+
+
 if __name__ == "__main__":
     app()