Skip to content

Companies

Companies(perfdb)

Class used for handling Companies. Can be accessed via perfdb.v_companies.

Parameters:

  • perfdb

    (PerfDB) –

    Top level object carrying all functionality and the connection handler.

Source code in echo_postgres/perfdb_root.py
Python
def __init__(self, perfdb: e_pg.PerfDB) -> None:
    """Base class that all subclasses should inherit from.

    Parameters
    ----------
    perfdb : PerfDB
        Top level object carrying all functionality and the connection handler.

    """
    self._perfdb: e_pg.PerfDB = perfdb

get(company_involved=None, company_responsible=None, company_types=None, output_type='dict')

Gets all companies with detailed information.

The most useful keys/columns returned are:

  • id
  • country

Parameters:

  • company_types

    (list[str] | None, default: None ) –

    Filter companies by their type name (column type_name in the view). By default None.

  • output_type

    (Literal['dict', 'DataFrame', 'pl.DataFrame'], default: 'dict' ) –

    Output type of the data. Can be one of ["dict", "DataFrame", "pl.DataFrame"] By default "dict"

Returns:

  • dict[str, dict[str, str | int]] –

    In case output_type is "dict", returns a dictionary in the format {name: {attribute: value, ...}, ...}

  • DataFrame –

    In case output_type is "DataFrame", returns a pandas DataFrame with the following format: index = name, columns = [attribute, ...]

  • DataFrame –

    In case output_type is "pl.DataFrame", returns a Polars DataFrame

Source code in echo_postgres/companies.py
Python
@validate_call
def get(
    self,
    company_involved: bool | None = None,
    company_responsible: bool | None = None,
    company_types: list[str] | None = None,
    output_type: Literal["dict", "DataFrame", "pl.DataFrame"] = "dict",
) -> dict[str, dict[str, str | int]] | pd.DataFrame | pl.DataFrame:
    """Gets all companies with detailed information.

    The most useful keys/columns returned are:

    - id
    - country

    Parameters
    ----------
    company_types : list[str] | None, optional
        Filter companies by their type name (column `type_name` in the view). By default None.
    output_type : Literal["dict", "DataFrame", "pl.DataFrame"], optional
        Output type of the data. Can be one of ["dict", "DataFrame", "pl.DataFrame"]
        By default "dict"

    Returns
    -------
    dict[str, dict[str, str | int]]
        In case output_type is "dict", returns a dictionary in the format {name: {attribute: value, ...}, ...}
    pd.DataFrame
        In case output_type is "DataFrame", returns a pandas DataFrame with the following format: index = name, columns = [attribute, ...]
    pl.DataFrame
        In case output_type is "pl.DataFrame", returns a Polars DataFrame
    """
    # checking arguments and building WHERE clause
    where = self._check_get_args(company_involved, company_responsible, company_types)

    query = sql.Composed([sql.SQL("SELECT * FROM performance.v_companies"), where, sql.SQL(" ORDER BY name")])

    df = self._perfdb.conn.read_to_polars(query)

    return convert_output(df, output_type)

get_ids(company_involved=None, company_responsible=None, company_types=None)

Gets all companies and their respective ids.

Parameters:

  • company_involved

    (bool | None, default: None ) –

    Filter by company_involved column when provided. By default None.

  • company_responsible

    (bool | None, default: None ) –

    Filter by company_responsible column when provided. By default None.

Returns:

  • dict[str, int] –

    Dictionary with all companies and their respective ids in the format {data_type: id, ...}.

Source code in echo_postgres/companies.py
Python
@validate_call
def get_ids(
    self,
    company_involved: bool | None = None,
    company_responsible: bool | None = None,
    company_types: list[str] | None = None,
) -> dict[str, int]:
    """Gets all companies and their respective ids.

    Parameters
    ----------
    company_involved : bool | None, optional
        Filter by `company_involved` column when provided. By default None.
    company_responsible : bool | None, optional
        Filter by `company_responsible` column when provided. By default None.

    Returns
    -------
    dict[str, int]
        Dictionary with all companies and their respective ids in the format {data_type: id, ...}.
    """
    # checking arguments and building WHERE clause
    where = self._check_get_args(company_involved, company_responsible, company_types)

    query = sql.Composed([sql.SQL("SELECT name, id FROM performance.v_companies"), where, sql.SQL(" ORDER BY name")])

    df = self._perfdb.conn.read_to_polars(query)

    return dict(zip(df["name"].to_list(), df["id"].to_list(), strict=False))