Database

Minoshiro uses caching to make search results faster and more accurate.

Build in database support

Minoshiro comes with built in support for SQLite3 and PostgreSQL databases.

To use the built in SQLite3 support, simply use the from_sqlite method as such:

from minoshiro import Minoshiro

async def main():

    db_path = 'path/to/database'

    robo = await Minoshiro.from_sqlite(db_path)

To use the built in PostgreSQL support, you will need asyncpg to be installed. Check Install for more information. Then, use the from_postgres method as such:

from minoshiro import Minoshiro


async def main():
    db_config = {
        "host": "localhost",
        "port": "5432",
        "user": "postgres",
        "database": "postgres"
    }

    robo = await Minoshiro.from_postgres(
        db_config, schema='my_schema'
    )

Extending DatabaseController

You may also write your custom implementation of the database controller if you wish. To get started, inherit from the DataController class as such:

from minoshiro import DataController


class MyDatabase(DataController):
    def __init__(self, logger):
        super().__init__(logger)

You will need to initialize the super class with a logger object.

Next, you will need to implement ALL of the following methods. The methods MUST be defined with async def.

@abstractmethod
async def get_identifier(self, query: str,
                         medium: Medium) -> Optional[Dict[Site, str]]:
    """
    Get the identifier of a given search query.

    :param query: the search query.
    :type query: str

    :param medium: the medium type.
    :type medium: Medium

    :return:
        A dict of all identifiers for this search query for all sites,
        None if nothing is found.
    :rtype: Optional[Dict[Site, str]]
    """
    raise NotImplementedError

@abstractmethod
async def set_identifier(self, name: str, medium: Medium,
                         site: Site, identifier: str):
    """
    Set the identifier for a given name.

    :param name: the name.
    :type name: str

    :param medium: the medium type.
    :type medium: Medium

    :param site: the site.
    :type site: Site

    :param identifier: the identifier.
    :type identifier: str
    """
    raise NotImplementedError

@abstractmethod
async def medium_data_by_id(self, id_: str, medium: Medium,
                            site: Site) -> Optional[dict]:
    """
    Get data by id.

    :param id_: the id.
    :type id_: str

    :param medium: the medium type.
    :type medium: Medium

    :param site: the site.
    :type site: Site

    :return: the data for that id if found.
    :rtype: Optional[dict]
    """
    raise NotImplementedError

@abstractmethod
async def set_medium_data(self, id_: str, medium: Medium,
                          site: Site, data: dict):
    """
    Set the data for a given id.

    :param id_: the id.
    :type id_: str

    :param medium: the medium type.
    :type medium: Medium

    :param site: the site.
    :type site: Site

    :param data: the data for the id.
    :type data: dict
    """
    raise NotImplementedError