API Reference

biokb_coconut.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_coconut/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_coconut.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_coconut/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_coconut.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_coconut/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 | None	SQLAlchemy engine URL (default: None, will use environment variable or default)	required
delete_files	bool	Delete downloaded source files after import (default: False)	required
env	str | None	Environment file to load for configuration (default: None)	required

Source code in src/biokb_coconut/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="Delete downloaded source files after import [default: False]",
)
@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 import_data(
    force_download: bool,
    connection_string: str | None,
    delete_files: bool,
    env: str | None,
) -> None:
    """Import data.

    Args:
        force_download (bool): Force re-download of the source file (default: False)
        connection_string (str | None): SQLAlchemy engine URL (default: None, will use environment variable or default)
        delete_files (bool): Delete downloaded source files after import (default: False)
        env (str | None): 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 | None	SQLAlchemy engine URL (default: None, will use environment variable or default)	required
env	str | None	Environment file to load for configuration (default: None)	required

Source code in src/biokb_coconut/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: str | None) -> None:
    """Create TTL files from local database.

    Args:
        connection_string (str | None): SQLAlchemy engine URL (default: None, will use environment variable or default)
        env (str | None): 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
        )  # Load environment variables from the specified .env file, override existing env variables if any
        connection_string = os.getenv("CONNECTION_STR")
        if connection_string is None:
            logger.warning(
                "CONNECTION_STR environment variable not found. Using default connection string."
            )
            connection_string = DB_DEFAULT_CONNECTION_STR
    if connection_string is None:
        logger.warning(
            "No connection string provided. Using default connection string."
        )
        connection_string = DB_DEFAULT_CONNECTION_STR

    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_coconut/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
env	str | None	Environment file to load for configuration (default: None)	required

Source code in src/biokb_coconut/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: 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 (str | None): 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
        )  # Load environment variables from the specified .env file, override existing env variables if any
    # 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)