API Reference

biokb_taxtree.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.	False

Returns:

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

Source code in src/biokb_taxtree/db/manager.py

def import_data(
    engine: Optional[Engine] = None,
    force_download: bool = False,
    delete_files: bool = False,
) -> 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_taxtree.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 src/biokb_taxtree/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_taxtree.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 src/biokb_taxtree/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.

Parameters:

Name	Type	Description	Default
force_download	bool	Force re-download of the source file (default: False)	required
connection_string	str	SQLAlchemy engine URL (default: sqlite:///taxtree.db)	required
delete_files	bool	Delete downloaded source files after import (default: False)	required
env	Optional[str]	Environment file to load for configuration (default: None)	None

Source code in src/biokb_taxtree/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(
    "-k",
    "--delete-files",
    is_flag=True,
    type=bool,
    default=False,
    help="Delete downloaded source files after import [default: False]",
)
@click.option(
    "-c",
    "--connection-string",
    type=str,
    default=None,
    help=f"SQLAlchemy engine URL [default: sqlite:///{PROJECT_NAME}.db]",
)
@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: Optional[str],
    delete_files: bool,
    env: Optional[str] = None,
) -> None:
    """Import data.

    Args:
        force_download (bool): Force re-download of the source file (default: False)
        connection_string (str): SQLAlchemy engine URL (default: sqlite:///taxtree.db)
        delete_files (bool): Delete downloaded source files after import (default: False)
        env (Optional[str]): Environment file to load for configuration (default: None)
    """
    if env:
        if connection_string:
            logger.warning(
                "Both environment file and connection string provided. Environment have priority."
            )
        if not os.path.exists(env):
            logger.error("Environment file %s not found.", env)
            return
        load_dotenv(env, override=True)
        connection_string = os.getenv("CONNECTION_STR")
        if connection_string is None:
            logger.warning(
                "CONNECTION_STR environment variable not found. Using default connection string."
            )

    engine: Engine | None = (
        create_engine(connection_string) if connection_string else None
    )
    DbManager(engine=engine).import_data(
        force_download=force_download, delete_files=delete_files
    )

create_ttls

Create TTL files from local database.

Parameters:

Name	Type	Description	Default
connection_string	str	SQLAlchemy engine URL (default: sqlite:///taxtree.db)	required

Source code in src/biokb_taxtree/cli.py

@main.command("create-ttls")
@click.option(
    "-c",
    "--connection-string",
    type=str,
    default=f"sqlite:///{PROJECT_NAME}.db",
    help=f"SQLAlchemy engine URL [default: sqlite:///{PROJECT_NAME}.db]",
)
def create_ttls(connection_string: str) -> None:
    """Create TTL files from local database.

    Args:
        connection_string (str): SQLAlchemy engine URL (default: sqlite:///taxtree.db)
    """
    path_to_zip = TurtleCreator(create_engine(connection_string)).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 src/biokb_taxtree/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	'0.0.0.0'
port	int	API server port	8000
user	str	API username	'admin'
password	str	API password	'admin'
env	Optional[str]	Environment file to load for configuration (default: None)	None

Source code in src/biokb_taxtree/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 = "0.0.0.0",
    port: int = 8000,
    user: str = "admin",
    password: str = "admin",
    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
        env (Optional[str]): Environment file to load for configuration (default: None)
    """
    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)