User Role Instances¶

`UserRoleInstances(perfdb)` ¶

Class used for handling User Roles. Can be accessed via perfdb.users.roles.instances.

Parameters:

perfdb ¶
(PerfDB) –

Top level object carrying all functionality and the connection handler.

Source code in echo_postgres/perfdb_root.py

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

`delete(names)` ¶

Deletes user roles from the database.

Parameters:

names ¶
(list[str]) –

List of user role names to delete.

Source code in echo_postgres/user_role_instances.py

@validate_call
def delete(self, names: list[str]) -> None:
    """Deletes user roles from the database.

    Parameters
    ----------
    names : list[str]
        List of user role names to delete.
    """
    if not names:
        return

    where_query = sql.SQL("WHERE name = ANY({user_names})").format(
        user_names=sql.Literal(names),
    )
    query = sql.SQL("DELETE FROM performance.user_roles {where_query}").format(
        where_query=where_query,
    )

    with self._perfdb.conn.reconnect() as conn:
        # deleting
        result = conn.execute(query)

    logger.debug(f"Deleted {result.rowcount} rows from the user_roles table")

`get(names=None, output_type='dict')` ¶

Gets all user roles with detailed information.

The most useful keys/columns returned are:

id
display_name
description
permissions, which is a list of dicts with the following keys:
- permission_type_id
- permission_type_name
- value
- permitted

Parameters:

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

List of role names to retrieve information for.

If None, retrieves information for all roles.
output_type ¶
(Literal['dict', 'DataFrame'], default: 'dict' ) –

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

Returns:

dict[str, dict[str, Any]] –

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

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

Source code in echo_postgres/user_role_instances.py

@validate_call
def get(
    self,
    names: list[str] | None = None,
    output_type: Literal["dict", "DataFrame"] = "dict",
) -> dict[str, dict[str, Any]] | DataFrame:
    """Gets all user roles with detailed information.

    The most useful keys/columns returned are:

    - id
    - display_name
    - description
    - permissions, which is a list of dicts with the following keys:
        - permission_type_id
        - permission_type_name
        - value
        - permitted

    Parameters
    ----------
    names : list[str] | None, optional
        List of role names to retrieve information for.

        If None, retrieves information for all roles.

    output_type : Literal["dict", "DataFrame"], optional
        Output type of the data. Can be one of ["dict", "DataFrame"]
        By default "dict"

    Returns
    -------
    dict[str, dict[str, Any]]
        In case output_type is "dict", returns a dictionary in the format {name: {attribute: value, ...}, ...}
    DataFrame
        In case output_type is "DataFrame", returns a DataFrame with the following format: index = name, columns = [attribute, ...]
    """
    where_query = (
        sql.SQL("WHERE name = ANY({user_names})").format(
            user_names=sql.Literal(names),
        )
        if names
        else sql.SQL("")
    )

    query = sql.SQL("SELECT * FROM performance.v_user_roles {where_query} ORDER BY name").format(
        where_query=where_query,
    )

    with self._perfdb.conn.reconnect() as conn:
        df = conn.read_to_pandas(query)
    df = df.set_index("name")

    return df.to_dict(orient="index") if output_type == "dict" else df

`get_ids(names=None)` ¶

Gets all user roles and their respective ids.

Parameters:

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

List of role names to retrieve ids for.

If None, retrieves ids for all user roles.

Returns:

dict[str, int] –

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

Source code in echo_postgres/user_role_instances.py

@validate_call
def get_ids(self, names: list[str] | None = None) -> dict[str, int]:
    """Gets all user roles and their respective ids.

    Parameters
    ----------
    names : list[str] | None, optional
        List of role names to retrieve ids for.

        If None, retrieves ids for all user roles.

    Returns
    -------
    dict[str, int]
        Dictionary with all user roles and their respective ids in the format {data_type: id, ...}.
    """
    where_query = (
        sql.SQL("WHERE name = ANY({user_names})").format(
            user_names=sql.Literal(names),
        )
        if names
        else sql.SQL("")
    )
    query = sql.SQL("SELECT name, id FROM performance.user_roles {where_query} ORDER BY name").format(
        where_query=where_query,
    )

    with self._perfdb.conn.reconnect() as conn:
        df = conn.read_to_pandas(query)

    return df.set_index("name").to_dict()["id"]

`insert(name, display_name=None, description=None, on_conflict='ignore')` ¶

Inserts or updates a user role into the database.

Fields left as None will not be inserted/updated.

Parameters:

name ¶
(str) –

Name of the user role.
display_name ¶
(str | None, default: None ) –

Display name of the user role, by default None.
description ¶
(str | None, default: None ) –

Description of the user role, by default None.
on_conflict ¶
(Literal['ignore', 'update'], default: 'ignore' ) –

What to do in case of conflict. Can be one of ["ignore", "update"]. By default "ignore"

Source code in echo_postgres/user_role_instances.py

@validate_call
def insert(
    self,
    name: str,
    display_name: str | None = None,
    description: str | None = None,
    on_conflict: Literal["ignore", "update"] = "ignore",
) -> None:
    """Inserts or updates a user role into the database.

    Fields left as None will not be inserted/updated.

    Parameters
    ----------
    name : str
        Name of the user role.
    display_name : str | None, optional
        Display name of the user role, by default None.
    description : str | None, optional
        Description of the user role, by default None.
    on_conflict : Literal["ignore", "update"], optional
        What to do in case of conflict. Can be one of ["ignore", "update"].
        By default "ignore"
    """
    # creating polars DataFrame with the data to insert
    data = {
        "name": [name],
    }
    if display_name is not None:
        data["display_name"] = [display_name]
    if description is not None:
        data["description"] = [description]

    df = pl.DataFrame(data)

    # inserting the data
    if_exists_mapping = {
        "ignore": "append",
        "update": "update",
    }
    with self._perfdb.conn.reconnect() as conn:
        conn.polars_to_sql(
            df=df,
            table_name="user_roles",
            schema="performance",
            if_exists=if_exists_mapping[on_conflict],
            conflict_cols=["name"],
        )

    logger.debug(f"Inserted/updated {len(df)} user roles into the database")

User Role Instances¶