comparison save_to_db.py @ 4:c7a7520afb4b draft

planemo upload for repository https://github.com/brsynth commit 6ae809b563b40bcdb6be2e74fe2a84ddad5484ae

3:eca0d710bbe9

import argparse
import socket
import os
import re
import pandas as pd
from sqlalchemy import create_engine, inspect
from sqlalchemy.sql import text
from sqlalchemy.engine.url import make_url
from sqlalchemy.exc import OperationalError

def fix_db_uri(uri):
    """Replace __at__ with @ in the URI if needed."""
    return uri.replace("__at__", "@")

    parser.add_argument("--db_uri", required=True, help="Database URI connection string")
    parser.add_argument("--table", required=True, help="Table name in the database")
    parser.add_argument("--fragment_column", required=True, help="Fragment column name in the database")
    parser.add_argument("--output", required=True, help="Text report")
    parser.add_argument("--file_name_mapping", required=True, help="real fragments names")
    args = parser.parse_args()

    # Start the Docker container (if not already running)
    gb_file_list = [f.strip() for f in args.input.split(",") if f.strip()]
    db_uri = fix_db_uri(args.db_uri)
    db_name = extract_db_name(db_uri)
    start_postgres_container(db_name)

    # Wait until the database is ready
    wait_for_db(db_uri)

    # Fetch annotations from the database and save as gb
    push_gb_annotations(gb_file_list, args.sequence_column, args.annotation_column, db_uri, args.table, args.fragment_column, args.output, args.file_name_mapping)

if __name__ == "__main__":
    main()

4:c7a7520afb4b

import argparse
import socket
import os
import re
import pandas as pd
import json
from sqlalchemy import create_engine, inspect
from sqlalchemy.sql import text
from sqlalchemy.engine.url import make_url
from sqlalchemy.exc import OperationalError

def resolve_parameters(user_params: dict, json_params: dict, keys: list):
    resolved = {}
    for key in keys:
        # Prefer user parameter if it's provided (not None or empty string)
        if key in user_params and user_params[key]:
            resolved[key] = user_params[key]
        else:
            resolved[key] = json_params.get(f"JSON_{key}")
    return resolved

def fix_db_uri(uri):
    """Replace __at__ with @ in the URI if needed."""
    return uri.replace("__at__", "@")

    parser.add_argument("--db_uri", required=True, help="Database URI connection string")
    parser.add_argument("--table", required=True, help="Table name in the database")
    parser.add_argument("--fragment_column", required=True, help="Fragment column name in the database")
    parser.add_argument("--output", required=True, help="Text report")
    parser.add_argument("--file_name_mapping", required=True, help="real fragments names")
    parser.add_argument("--json_conf", required=False, help="JSON config file with DB parameters")
    args = parser.parse_args()

    # Load JSON config if provided
    json_config = {}
    if args.json_conf != 'None' or '':
        with open(args.json_conf, "r") as f:
            json_config = json.load(f)

    # Prefer user input; fallback to JSON_ values if not provided
    user_params = {
        "table": args.table,
        "sequence_column": args.sequence_column,
        "annotation_column": args.annotation_column,
        "fragment_column": args.fragment_column,
        "db_uri": args.db_uri
    }

    keys = ["table", "sequence_column", "annotation_column", "fragment_column", "db_uri"]
    resolved = resolve_parameters(user_params, json_config, keys)

    # Unpack resolved parameters
    table = resolved["table"]
    sequence_column = resolved["sequence_column"]
    annotation_column = resolved["annotation_column"]
    fragment_column = resolved["fragment_column"]
    db_uri = fix_db_uri(resolved["db_uri"])

    # Prepare gb files
    gb_file_list = [f.strip() for f in args.input.split(",") if f.strip()]

    # Start and wait for DB
    db_name = extract_db_name(db_uri)
    start_postgres_container(db_name)
    wait_for_db(db_uri)

    # Push annotations
    push_gb_annotations(
        gb_file_list,
        sequence_column,
        annotation_column,
        db_uri,
        table,
        fragment_column,
        args.output,
        args.file_name_mapping
    )

if __name__ == "__main__":
    main()