core_server

ensembl.io.genomio.database.core_server

Interface to a Mysql server with core databases.

CoreServer

Basic interface to a MySQL server with core databases.

Allows to get a filtered list of databases.

Source code in src/python/ensembl/io/genomio/database/core_server.py

class CoreServer:
    """Basic interface to a MySQL server with core databases.

    Allows to get a filtered list of databases.
    """

    def __init__(self, server_url: URL) -> None:
        logging.debug(f"Connect to {server_url}")
        self.engine = sqlalchemy.create_engine(server_url)

    def get_all_core_names(self) -> List[str]:
        """Query the server and retrieve all database names that look like Ensembl cores."""

        with self.engine.connect() as connection:
            all_query = connection.execute(text(r"SHOW DATABASES LIKE '%%_core_%%'"))
            dbs = [row[0] for row in all_query.fetchall()]
        logging.info(f"{len(dbs)} core databases on the server")
        return dbs

    def get_cores(
        self,
        *,
        prefix: str = "",
        build: Optional[int] = None,
        version: Optional[int] = None,
        dbname_re: str = "",
        db_list: Optional[List[str]] = None,
    ) -> List[str]:
        """Returns a list of core databases, filtered if requested.

        Args:
            prefix: Filter by prefix (no "_" is added automatically).
            build: Filter by VEuPathDB build number.
            version: Filter by Ensembl version.
            dbname_re: Filter by dbname regular expression.
            db_list: Explicit list of database names.
        """
        dbs = []

        dbs = self.get_all_core_names()

        # Check if there are databases returned from query to host
        if not dbs:
            logging.warning("No databases returned from query")

        if db_list:
            logging.debug(f"Filter with db list: {db_list}")
            dbs = [db for db in dbs if db in db_list]
        if prefix:
            dbs = [db for db in dbs if db.startswith(f"{prefix}")]
        if dbname_re:
            dbname_m = re.compile(dbname_re)
            dbs = list(filter(dbname_m.search, dbs))
        if build is not None:
            dbs = [db for db in dbs if re.search(rf"_core_{build}_\d+_\d+$", db)]
        if version is not None:
            dbs = [db for db in dbs if re.search(rf"_core_\d+_{version}_\d+$", db)]

        logging.info(f"{len(dbs)} core databases remain after filtering")

        return dbs

engine = sqlalchemy.create_engine(server_url) instance-attribute

get_all_core_names()

Query the server and retrieve all database names that look like Ensembl cores.

Source code in src/python/ensembl/io/genomio/database/core_server.py

def get_all_core_names(self) -> List[str]:
    """Query the server and retrieve all database names that look like Ensembl cores."""

    with self.engine.connect() as connection:
        all_query = connection.execute(text(r"SHOW DATABASES LIKE '%%_core_%%'"))
        dbs = [row[0] for row in all_query.fetchall()]
    logging.info(f"{len(dbs)} core databases on the server")
    return dbs

get_cores(*, prefix='', build=None, version=None, dbname_re='', db_list=None)

Returns a list of core databases, filtered if requested.

Parameters:

Name	Type	Description	Default
prefix	str	Filter by prefix (no "_" is added automatically).	''
build	Optional[int]	Filter by VEuPathDB build number.	None
version	Optional[int]	Filter by Ensembl version.	None
dbname_re	str	Filter by dbname regular expression.	''
db_list	Optional[List[str]]	Explicit list of database names.	None

Source code in src/python/ensembl/io/genomio/database/core_server.py

def get_cores(
    self,
    *,
    prefix: str = "",
    build: Optional[int] = None,
    version: Optional[int] = None,
    dbname_re: str = "",
    db_list: Optional[List[str]] = None,
) -> List[str]:
    """Returns a list of core databases, filtered if requested.

    Args:
        prefix: Filter by prefix (no "_" is added automatically).
        build: Filter by VEuPathDB build number.
        version: Filter by Ensembl version.
        dbname_re: Filter by dbname regular expression.
        db_list: Explicit list of database names.
    """
    dbs = []

    dbs = self.get_all_core_names()

    # Check if there are databases returned from query to host
    if not dbs:
        logging.warning("No databases returned from query")

    if db_list:
        logging.debug(f"Filter with db list: {db_list}")
        dbs = [db for db in dbs if db in db_list]
    if prefix:
        dbs = [db for db in dbs if db.startswith(f"{prefix}")]
    if dbname_re:
        dbname_m = re.compile(dbname_re)
        dbs = list(filter(dbname_m.search, dbs))
    if build is not None:
        dbs = [db for db in dbs if re.search(rf"_core_{build}_\d+_\d+$", db)]
    if version is not None:
        dbs = [db for db in dbs if re.search(rf"_core_\d+_{version}_\d+$", db)]

    logging.info(f"{len(dbs)} core databases remain after filtering")

    return dbs