Utilities

Little blurb

`minall.utils.database`

Utilities to manage SQLite database connection.

The module contains the following function and class:

connect_to_database(database) - If provided with path, connects to embedded SQLite database; otherwise, connects to in-memory SQLite database.
SQLiteWrapper(connection) - Stores connection and cursor, executes queries.

`SQLiteWrapper`

Class to store SQLite database connection and execute SQL queries.

Examples:

>>> wrapper = SQLiteWrapper(connection=connect_to_database())
>>> _ = wrapper(query="create table test(name text)")
>>> wrapper.select("select * from test")
[]

Source code in minall/utils/database.py

class SQLiteWrapper:
    """Class to store SQLite database connection and execute SQL queries.

    Examples:
        >>> wrapper = SQLiteWrapper(connection=connect_to_database())
        >>> _ = wrapper(query="create table test(name text)")
        >>> wrapper.select("select * from test")
        []

    """

    def __init__(self, connection: Connection) -> None:
        """Store database connection and create cursor.

        Examples:
            >>> wrapper = SQLiteWrapper(connection=connect_to_database())
            >>> type(wrapper.cursor)
            <class 'sqlite3.Cursor'>

        Args:
            connection (Connection): Connection to SQLite database.
        """
        self.connection = connection
        self.cursor = self.connection.cursor()

    def __call__(self, query: str, values: List[Tuple] | None = None) -> None:
        """Execute and commit SQL query.

        Examples:
            >>> wrapper = SQLiteWrapper(connection=connect_to_database())
            >>> _ = wrapper(query="create table test(name text)")
            >>> wrapper.select("select * from test")
            []

        Args:
            query (str): Query string, can contain SQL place holders for values (?).
            values (list[tuple] | None, optional): Values to be included in query. Defaults to None.

        Raises:
            Exception: `sqlite3` Exception caused either by falling to execute query with cursor or by failing to commit changes to connected database.
        """
        try:
            if values:
                self.cursor.execute(query, values)
            else:
                self.cursor.execute(query)
            self.connection.commit()
        except Exception as e:
            print("\n", query, "\n")
            print(values)
            raise e

    def select(self, query: str) -> List:
        """Return selection from SQLite database.

        Examples:
            >>> wrapper = SQLiteWrapper(connection=connect_to_database())
            >>> _ = wrapper(query="create table test(name text)")
            >>> wrapper.select("select * from test")
            []

        Args:
            query (str): SQL select query.

        Raises:
            Exception: `sqlite3` Exception caused either by falling to execute query with cursor or by failing to commit changes to connected database.

        Returns:
            List: Array of results from SQL query.
        """
        try:
            response = self.cursor.execute(query)
            self.connection.commit()
        except Exception as e:
            print("\n", query, "\n")
            raise e
        else:
            return response.fetchall()

`call(query, values=None)`

Execute and commit SQL query.

Examples:

>>> wrapper = SQLiteWrapper(connection=connect_to_database())
>>> _ = wrapper(query="create table test(name text)")
>>> wrapper.select("select * from test")
[]

Parameters:

Name	Type	Description	Default
`query`	`str`	Query string, can contain SQL place holders for values (?).	required
`values`	`list[tuple] \| None`	Values to be included in query. Defaults to None.	`None`

Raises:

Type	Description
`Exception`	`sqlite3` Exception caused either by falling to execute query with cursor or by failing to commit changes to connected database.

Source code in minall/utils/database.py

def __call__(self, query: str, values: List[Tuple] | None = None) -> None:
    """Execute and commit SQL query.

    Examples:
        >>> wrapper = SQLiteWrapper(connection=connect_to_database())
        >>> _ = wrapper(query="create table test(name text)")
        >>> wrapper.select("select * from test")
        []

    Args:
        query (str): Query string, can contain SQL place holders for values (?).
        values (list[tuple] | None, optional): Values to be included in query. Defaults to None.

    Raises:
        Exception: `sqlite3` Exception caused either by falling to execute query with cursor or by failing to commit changes to connected database.
    """
    try:
        if values:
            self.cursor.execute(query, values)
        else:
            self.cursor.execute(query)
        self.connection.commit()
    except Exception as e:
        print("\n", query, "\n")
        print(values)
        raise e

`init(connection)`

Store database connection and create cursor.

Examples:

>>> wrapper = SQLiteWrapper(connection=connect_to_database())
>>> type(wrapper.cursor)
<class 'sqlite3.Cursor'>

Parameters:

Name	Type	Description	Default
`connection`	`Connection`	Connection to SQLite database.	required

Source code in minall/utils/database.py

def __init__(self, connection: Connection) -> None:
    """Store database connection and create cursor.

    Examples:
        >>> wrapper = SQLiteWrapper(connection=connect_to_database())
        >>> type(wrapper.cursor)
        <class 'sqlite3.Cursor'>

    Args:
        connection (Connection): Connection to SQLite database.
    """
    self.connection = connection
    self.cursor = self.connection.cursor()

`select(query)`

Return selection from SQLite database.

Examples:

>>> wrapper = SQLiteWrapper(connection=connect_to_database())
>>> _ = wrapper(query="create table test(name text)")
>>> wrapper.select("select * from test")
[]

Parameters:

Name	Type	Description	Default
`query`	`str`	SQL select query.	required

Raises:

Type	Description
`Exception`	`sqlite3` Exception caused either by falling to execute query with cursor or by failing to commit changes to connected database.

Returns:

Name	Type	Description
`List`	`List`	Array of results from SQL query.

Source code in minall/utils/database.py

def select(self, query: str) -> List:
    """Return selection from SQLite database.

    Examples:
        >>> wrapper = SQLiteWrapper(connection=connect_to_database())
        >>> _ = wrapper(query="create table test(name text)")
        >>> wrapper.select("select * from test")
        []

    Args:
        query (str): SQL select query.

    Raises:
        Exception: `sqlite3` Exception caused either by falling to execute query with cursor or by failing to commit changes to connected database.

    Returns:
        List: Array of results from SQL query.
    """
    try:
        response = self.cursor.execute(query)
        self.connection.commit()
    except Exception as e:
        print("\n", query, "\n")
        raise e
    else:
        return response.fetchall()

`connect_to_database(database=None)`

Connect to SQLite database.

Examples:

>>> conn = connect_to_database()
>>> type(conn)
<class 'sqlite3.Connection'>
>>> _ = conn.cursor().execute("create table test(name text)")
>>> conn.cursor().execute("select * from test").fetchall()
[]

Parameters:

Name	Type	Description	Default
`database`	`str \| None`	If given, path to embedded SQLite database. Defaults to None.	`None`

Returns:

Name	Type	Description
`Connection`	`Connection`	description

Source code in minall/utils/database.py

def connect_to_database(database: str | None = None) -> Connection:
    """Connect to SQLite database.

    Examples:
        >>> conn = connect_to_database()
        >>> type(conn)
        <class 'sqlite3.Connection'>
        >>> _ = conn.cursor().execute("create table test(name text)")
        >>> conn.cursor().execute("select * from test").fetchall()
        []

    Args:
        database (str | None, optional): If given, path to embedded SQLite database. Defaults to None.

    Returns:
        Connection: _description_
    """

    if database:
        [p.mkdir(exist_ok=True) for p in Path(database).parents]
        connection = sqlite3.connect(database)
    else:
        connection = sqlite3.connect(":memory:")
    return connection

`minall.utils.parse_config`

Data class to store and manage minet client credentials.

The class APIKeys contains the following methods and properties:

__init__(config) - Parses the minet client configuration details.
env_string() - Formats the minet client credentials as an environment variable string.
load_config_file(config_file) - Parse client configuration details from JSON or YAML file.

`APIKeys` `dataclass`

Data class to store and manage minet client credentials.

Attributes:

Name	Type	Description
`buzzsumo_token`	`Optional[str]`	Buzzsumo API token. Optional.
`crowdtangle_token`	`Optional[str]`	CrowdTangle API token. Optional.
`crowdtangle_rate_limit`	`Optional[str]`	CrowdTangle API rate limit, cast as a string. Optional.
`youtube_key`	`Optional[List[str]])`	List of YouTube API keys. Optional.

Source code in minall/utils/parse_config.py

@dataclass
class APIKeys:
    """Data class to store and manage minet client credentials.

    Attributes:
        buzzsumo_token (Optional[str]): Buzzsumo API token. Optional.
        crowdtangle_token (Optional[str]):  CrowdTangle API token. Optional.
        crowdtangle_rate_limit (Optional[str]): CrowdTangle API rate limit, cast as a string. Optional.
        youtube_key (Optional[List[str]]) : List of YouTube API keys. Optional.
    """

    buzzsumo_token: Optional[str]
    crowdtangle_token: Optional[str]
    crowdtangle_rate_limit: Optional[str]
    youtube_key: Optional[List[str]]

    def __init__(self, config: str | dict | None = None):
        """Parse and save minet API client configuration details.

        Examples:
            >>> keys = APIKeys(config={"youtube": {"key": "key1,key2"}})
            >>> keys
            APIKeys(buzzsumo_token=None, crowdtangle_token=None, crowdtangle_rate_limit=None, youtube_key=["key1", "key2"])
            >>> keys.youtube_key
            ["key1, "key2]

        Args:
            config (str | dict | None, optional): If string, string is treated like file path to JSON or YAML file that contains details; if dict, details are directly parsed; if None, details are searched from environment variables. Defaults to None.
        """
        if config:
            if isinstance(config, str):
                parsed_config = self.load_config_file(config)
            else:
                parsed_config = config
            self.buzzsumo_token = parsed_config["buzzsumo"]["token"]
            self.crowdtangle_token = parsed_config["crowdtangle"]["token"]
            self.crowdtangle_rate_limit = parsed_config["crowdtangle"]["rate_limit"]
            yt_keys = parsed_config["youtube"]["key"]
            if isinstance(yt_keys, list):
                self.youtube_key = yt_keys
            else:
                self.youtube_key = parsed_config["youtube"]["key"].split(",")
        else:
            self.buzzsumo_token = os.environ.get("BUZZSUMO_TOKEN")
            self.crowdtangle_token = os.environ.get("CROWDTANGLE_TOKEN")
            self.crowdtangle_rate_limit = os.environ.get("CROWDTANGLE_RATE_LIMIT")
            youtube_key = os.environ.get("YOUTUBE_KEY")
            if youtube_key:
                self.youtube_key = youtube_key.split(",")
            else:
                self.youtube_key = []

    @property
    def env_string(self) -> str:
        r"""Formatted string for setting environment variables.

        Examples:
            >>> keys = APIKeys(config={'buzzsumo': {'token': 'bz_token'}, 'crowdtangle': {'token': 'ct_token', 'rate_limit': 10}, 'youtube': {'key': 'key1,key2'}})
            >>> keys.env_string
            "BUZZSUMO_TOKEN=bz_token\nCROWDTANGLE_TOKEN=ct_token\nCROWDTANGLE_RATE_LIMIT=10\nYOUTUBE_KEY=['key1', 'key2']\n"

        Returns:
            str: String declaring environment variables.
        """

        return "BUZZSUMO_TOKEN={bz}\nCROWDTANGLE_TOKEN={ct}\nCROWDTANGLE_RATE_LIMIT={crl}\nYOUTUBE_KEY={yt}\n".format(
            bz=self.buzzsumo_token,
            ct=self.crowdtangle_token,
            crl=self.crowdtangle_rate_limit,
            yt=self.youtube_key,
        )

    def load_config_file(self, config_file: str) -> dict:
        """Parse dictionary from JSON or YAML configuration file.

        Args:
            config_file (str): Path to JSON or YAML file.

        Raises:
            OSError: Error raised if given file path does not have the extension ".json", ".yml", or ".yaml".

        Returns:
            dict: Parsed dictionary from JSON or YAML configuration file.
        """

        with open(config_file) as f:
            extension = Path(config_file).suffix
            if extension == ".json":
                return json.load(f)
            elif extension == ".yml" or extension == ".yaml":
                return yaml.safe_load(f)
            else:
                raise OSError

`env_string: str` `property`

Formatted string for setting environment variables.

Examples:

>>> keys = APIKeys(config={'buzzsumo': {'token': 'bz_token'}, 'crowdtangle': {'token': 'ct_token', 'rate_limit': 10}, 'youtube': {'key': 'key1,key2'}})
>>> keys.env_string
"BUZZSUMO_TOKEN=bz_token\nCROWDTANGLE_TOKEN=ct_token\nCROWDTANGLE_RATE_LIMIT=10\nYOUTUBE_KEY=['key1', 'key2']\n"

Returns:

Name	Type	Description
`str`	`str`	String declaring environment variables.

`init(config=None)`

Parse and save minet API client configuration details.

Examples:

>>> keys = APIKeys(config={"youtube": {"key": "key1,key2"}})
>>> keys
APIKeys(buzzsumo_token=None, crowdtangle_token=None, crowdtangle_rate_limit=None, youtube_key=["key1", "key2"])
>>> keys.youtube_key
["key1, "key2]

Parameters:

Name	Type	Description	Default
`config`	`str \| dict \| None`	If string, string is treated like file path to JSON or YAML file that contains details; if dict, details are directly parsed; if None, details are searched from environment variables. Defaults to None.	`None`

Source code in minall/utils/parse_config.py

def __init__(self, config: str | dict | None = None):
    """Parse and save minet API client configuration details.

    Examples:
        >>> keys = APIKeys(config={"youtube": {"key": "key1,key2"}})
        >>> keys
        APIKeys(buzzsumo_token=None, crowdtangle_token=None, crowdtangle_rate_limit=None, youtube_key=["key1", "key2"])
        >>> keys.youtube_key
        ["key1, "key2]

    Args:
        config (str | dict | None, optional): If string, string is treated like file path to JSON or YAML file that contains details; if dict, details are directly parsed; if None, details are searched from environment variables. Defaults to None.
    """
    if config:
        if isinstance(config, str):
            parsed_config = self.load_config_file(config)
        else:
            parsed_config = config
        self.buzzsumo_token = parsed_config["buzzsumo"]["token"]
        self.crowdtangle_token = parsed_config["crowdtangle"]["token"]
        self.crowdtangle_rate_limit = parsed_config["crowdtangle"]["rate_limit"]
        yt_keys = parsed_config["youtube"]["key"]
        if isinstance(yt_keys, list):
            self.youtube_key = yt_keys
        else:
            self.youtube_key = parsed_config["youtube"]["key"].split(",")
    else:
        self.buzzsumo_token = os.environ.get("BUZZSUMO_TOKEN")
        self.crowdtangle_token = os.environ.get("CROWDTANGLE_TOKEN")
        self.crowdtangle_rate_limit = os.environ.get("CROWDTANGLE_RATE_LIMIT")
        youtube_key = os.environ.get("YOUTUBE_KEY")
        if youtube_key:
            self.youtube_key = youtube_key.split(",")
        else:
            self.youtube_key = []

`load_config_file(config_file)`

Parse dictionary from JSON or YAML configuration file.

Parameters:

Name	Type	Description	Default
`config_file`	`str`	Path to JSON or YAML file.	required

Raises:

Type	Description
`OSError`	Error raised if given file path does not have the extension ".json", ".yml", or ".yaml".

Returns:

Name	Type	Description
`dict`	`dict`	Parsed dictionary from JSON or YAML configuration file.

Source code in minall/utils/parse_config.py

def load_config_file(self, config_file: str) -> dict:
    """Parse dictionary from JSON or YAML configuration file.

    Args:
        config_file (str): Path to JSON or YAML file.

    Raises:
        OSError: Error raised if given file path does not have the extension ".json", ".yml", or ".yaml".

    Returns:
        dict: Parsed dictionary from JSON or YAML configuration file.
    """

    with open(config_file) as f:
        extension = Path(config_file).suffix
        if extension == ".json":
            return json.load(f)
        elif extension == ".yml" or extension == ".yaml":
            return yaml.safe_load(f)
        else:
            raise OSError

`minall.utils.progress_bar`

Context for rich progress bar.

`progress_bar()`

Rich progress bar with Spinner column.

Yields:

Type	Description
`Progress`	Generator[Progress, None, None]: Rich progress bar context

Source code in minall/utils/progress_bar.py

@contextmanager
def progress_bar() -> Generator[Progress, None, None]:
    """Rich progress bar with Spinner column.

    Yields:
        Generator[Progress, None, None]: Rich progress bar context
    """
    with Progress(
        TextColumn("[progress.description]{task.description}"),
        SpinnerColumn(),
        MofNCompleteColumn(),
        TimeElapsedColumn(),
    ) as progress:
        yield progress

Utilities

minall.utils.database

SQLiteWrapper

__call__(query, values=None)

__init__(connection)

select(query)

connect_to_database(database=None)

minall.utils.parse_config

APIKeys dataclass

env_string: str property

__init__(config=None)

load_config_file(config_file)

minall.utils.progress_bar

progress_bar()

`minall.utils.database`

`SQLiteWrapper`

`call(query, values=None)`

`init(connection)`

`select(query)`

`connect_to_database(database=None)`

`minall.utils.parse_config`

`APIKeys` `dataclass`

`env_string: str` `property`

`init(config=None)`

`load_config_file(config_file)`

`minall.utils.progress_bar`

`progress_bar()`