API Reference

biokb_wcvp.import_data

Import all data in database.

Parameters:

Name	Type	Description	Default
engine	Optional[Engine]	SQLAlchemy engine. Defaults to None.	None
force_download	bool	If True, will force download the data, even if files already exist. If False, it will skip the downloading part if files already exist locally. Defaults to False.	False
delete_files	bool	If True, downloaded files are deleted after import. Defaults to False.	True

Returns:

Type	Description
dict[str, int]	dict[str, int]: table=key and number of inserted=value

Source code in biokb_wcvp/db/manager.py

def import_data(
    engine: Optional[Engine] = None,
    force_download: bool = False,
    delete_files: bool = True,
) -> dict[str, int]:
    """Import all data in database.

    Args:
        engine (Optional[Engine]): SQLAlchemy engine. Defaults to None.
        force_download (bool, optional): If True, will force download the data, even if
            files already exist. If False, it will skip the downloading part if files
            already exist locally. Defaults to False.
        delete_files (bool, optional): If True, downloaded files are deleted after import.
            Defaults to False.

    Returns:
        dict[str, int]: table=key and number of inserted=value
    """
    db_manager = DbManager(engine)
    return db_manager.import_data(
        force_download=force_download, delete_files=delete_files
    )

biokb_wcvp.create_ttls

Create all turtle files.

If engine=None tries to get the settings from config ini file

If export_to_folder=None takes the default path.

Parameters:

Name	Type	Description	Default
engine	Engine | None	SQLAlchemy class. Defaults to None.	None
export_to_folder	str | None	Folder to export ttl files. Defaults to None.	None

Returns:

Name	Type	Description
str	str	path zipped file with ttls.

Source code in biokb_wcvp/rdf/turtle.py

def create_ttls(
    engine: Optional[Engine] = None,
    export_to_folder: Optional[str] = None,
) -> str:
    """Create all turtle files.

    If engine=None tries to get the settings from config ini file

    If export_to_folder=None takes the default path.

    Args:
        engine (Engine | None, optional): SQLAlchemy class. Defaults to None.
        export_to_folder (str | None, optional): Folder to export ttl files.
            Defaults to None.

    Returns:
        str: path zipped file with ttls.
    """
    ttl_creator = TurtleCreator(engine=engine)
    if export_to_folder:
        ttl_creator._set_ttls_folder(export_to_folder)
    return ttl_creator.create_ttls()

biokb_wcvp.import_ttls

Import data into Neo4J from zipped turtle files.

Parameters:

Name	Type	Description	Default
neo4j_uri	str | None	URI of the Neo4j database.	None
neo4j_user	str | None	Username for Neo4j.	None
neo4j_pwd	str | None	Password for Neo4j.	None
delete_existing_graph	bool	delete existing graph before import.	True

Returns: bool: True if import is successful.

Source code in biokb_wcvp/rdf/neo4j_importer.py

def import_ttls(
    neo4j_uri: str | None = None,
    neo4j_user: str | None = None,
    neo4j_pwd: str | None = None,
    delete_existing_graph: bool = True,
) -> bool:
    """Import data into Neo4J from zipped turtle files.

    Args:
        neo4j_uri (str | None): URI of the Neo4j database.
        neo4j_user (str | None): Username for Neo4j.
        neo4j_pwd (str | None): Password for Neo4j.
        delete_existing_graph (bool): delete existing graph before import.
    Returns:
        bool: True if import is successful.
    """
    importer = Neo4jImporter(
        neo4j_uri=neo4j_uri, neo4j_user=neo4j_user, neo4j_pwd=neo4j_pwd
    )
    result: bool = importer.import_ttls(delete_existing_graph=delete_existing_graph)
    return result

Command Line Interface (CLI)

import_data

Import data.

Source code in biokb_wcvp/cli.py

@main.command("import-data")
@click.option(
    "-f",
    "--force-download",
    is_flag=True,
    type=bool,
    default=False,
    help="Force re-download of the source file [default: False]",
)
@click.option(
    "-d",
    "--delete-files",
    is_flag=True,
    type=bool,
    default=False,
    help="Keep downloaded source files after import [default: False]",
)
@click.option(
    "-c",
    "--connection-string",
    type=str,
    default=None,
    help=f"SQLAlchemy engine URL [default: [default: {DB_DEFAULT_CONNECTION_STR}]",
)
@click.option(
    "-e",
    "--env",
    type=str,
    default=None,
    help="Environment file to load for configuration (default: None)",
)
def import_data(
    force_download: bool,
    connection_string: str | None,
    delete_files: bool,
    env: Optional[str] = None,
):
    """Import data."""
    try:
        engine = get_engine(connection_string=connection_string, env=env)
    except ValueError as e:
        logger.error("Error creating database engine: %s", e)
        return
    DbManager(engine=engine).import_data(
        force_download=force_download, delete_files=delete_files
    )

create_ttls

Create TTL files from local database.

Source code in biokb_wcvp/cli.py

@main.command("create-ttls")
@click.option(
    "-c",
    "--connection-string",
    type=str,
    default=None,
    help=f"SQLAlchemy engine URL [default: {DB_DEFAULT_CONNECTION_STR}]",
)
@click.option(
    "-e",
    "--env",
    type=str,
    default=None,
    help="Environment file to load for configuration (default: None)",
)
def create_ttls(connection_string: str | None, env: Optional[str] = None) -> None:
    """Create TTL files from local database."""
    try:
        engine = get_engine(connection_string=connection_string, env=env)
    except ValueError as e:
        logger.error("Error creating database engine: %s", e)
        return
    path_to_zip = TurtleCreator(engine).create_ttls()
    click.echo(
        f"Path to the zip file containing all generated Turtle files. {path_to_zip}"
    )

import_neo4j

Import TTL files into Neo4j database.

Source code in biokb_wcvp/cli.py

@main.command("import-neo4j")
@click.option(
    "--uri",
    "-i",
    default=neo4j_uri,
    help=f'Neo4j database URI [default:"{neo4j_uri}"]',
)
@click.option(
    "--user", "-u", default=neo4j_user, help=f'Neo4j username [default="{neo4j_user}"]'
)
@click.option("--password", "-p", default=None, help="Neo4j password")
def import_neo4j(uri: str, user: str, password: Optional[str]) -> None:
    """Import TTL files into Neo4j database."""
    if password is None:
        password = click.prompt(
            "Please enter the Neo4j password (input will be hidden)", hide_input=True
        )
    else:
        click.echo(
            "It is not recommended to provide the Neo4j password via command line."
        )
    Neo4jImporter(neo4j_uri=uri, neo4j_user=user, neo4j_pwd=password).import_ttls()

run_server

Run the API server.

Parameters:

Name	Type	Description	Default
host	str	API server host	required
port	int	API server port	required
user	str	API username	required
password	str	API password	required

Source code in biokb_wcvp/cli.py

@main.command("run-server")
@click.option(
    "--host", "-h", default="0.0.0.0", help="API server host [default: 0.0.0.0]"
)
@click.option("--port", "-P", default=8000, help="API server port [default: 8000]")
@click.option("--user", "-u", default="admin", help="API username [default=admin]")
@click.option("--password", "-p", default="admin", help="API password [default: admin]")
@click.option(
    "-e",
    "--env",
    type=str,
    default=None,
    help="Environment file to load for configuration (default: None)",
)
def run_server(
    host: str,
    port: int,
    user: str,
    password: str,
    env: Optional[str] = None,
) -> None:
    """Run the API server.

    Args:
        host (str): API server host
        port (int): API server port
        user (str): API username
        password (str): API password
    """
    # set env variables for API authentication
    if env:
        if not os.path.exists(env):
            logger.error("Environment file %s not found.", env)
            return
        load_dotenv(env, override=True)
    else:
        # set env variables for API authentication
        os.environ["API_USER"] = user
        os.environ["API_PASSWORD"] = password

    host_shown = "127.0.0.1" if host == "0.0.0.0" else host
    click.echo(f"API server running at http://{host_shown}:{port}/docs#/")
    run_api(host=host, port=port)