Settings

Settings(perfdb)

Class used for handling database settings. Can be accessed via perfdb.settings.

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

delete(section, key)

Deletes a setting from the database

Parameters:

• section

  (str) –

  Section name

• key

  (str) –

  Setting name

Returns:

• None –

Source code in echo_postgres/settings.py

Python

@validate_call
def delete(self, section: str, key: str) -> None:
    """Deletes a setting from the database

    Parameters
    ----------
    section : str
        Section name
    key : str
        Setting name

    Returns
    -------
    None
    """
    query = sql.SQL("""
        DELETE FROM performance.settings WHERE section = {section} AND key = {key}
    """).format(
        section=sql.Literal(section),
        key=sql.Literal(key),
    )

    self._perfdb.conn.execute(query)

get(sections=None, keys=None, output_type='dict')

Gets all database settings

The most useful keys/columns returned are:

• id: Unique identifier for the setting
• section: Section name
• key: Setting name
• value: Value of the setting (is a dictionary)
• description: Description of the setting
• schema: Schema used to validate if the setting value is correct, only for database internal checks
• modified_date: Date of the last modification of the setting

Parameters:

• sections

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

  List of sections to filter the settings by. If None, all sections will be returned. By default None

• keys

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

  List of keys to filter the settings by. If None, all keys will be returned. 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, dict[str, Any]]] –

  In case output_type is "dict", returns a dictionary in the format {section: {key: {"id": id, "value": value, "description": description, "schema": schema}}, ...}

• DataFrame –

  In case output_type is "DataFrame", returns a pandas DataFrame with the following format: MultiIndex = [section, key], columns = ["id", "value", "description", "schema"]

• DataFrame –

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

Source code in echo_postgres/settings.py

Python

@validate_call
def get(
    self,
    sections: list[str] | None = None,
    keys: list[str] | None = None,
    output_type: Literal["dict", "DataFrame", "pl.DataFrame"] = "dict",
) -> dict[str, dict[str, dict[str, Any]]] | pd.DataFrame | pl.DataFrame:
    """Gets all database settings

    The most useful keys/columns returned are:

    - id: Unique identifier for the setting
    - section: Section name
    - key: Setting name
    - value: Value of the setting (is a dictionary)
    - description: Description of the setting
    - schema: Schema used to validate if the setting value is correct, only for database internal checks
    - modified_date: Date of the last modification of the setting

    Parameters
    ----------
    sections : list[str] | None, optional
        List of sections to filter the settings by. If None, all sections will be returned.
        By default None
    keys : list[str] | None, optional
        List of keys to filter the settings by. If None, all keys will be returned.
        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, dict[str, Any]]]
        In case output_type is "dict", returns a dictionary in the format {section: {key: {"id": id, "value": value, "description": description, "schema": schema}}, ...}
    pd.DataFrame
        In case output_type is "DataFrame", returns a pandas DataFrame with the following format: MultiIndex = [section, key], columns = ["id", "value", "description", "schema"]
    pl.DataFrame
        In case output_type is "pl.DataFrame", returns a Polars DataFrame
    """
    where = []
    if sections is not None:
        where.append(
            sql.SQL("section = ANY({sections})").format(
                sections=sql.Literal(sections),
            ),
        )
    if keys is not None:
        where.append(
            sql.SQL("key = ANY({keys})").format(
                keys=sql.Literal(keys),
            ),
        )

    where = sql.Composed([sql.SQL("WHERE "), sql.SQL(" AND ").join(where)]) if where else sql.SQL("")

    query = sql.SQL(
        "SELECT section, key, value::TEXT, description, schema::TEXT, modified_date, id FROM performance.settings {where} ORDER BY section, key",
    ).format(
        where=where,
    )

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

    if output_type == "pl.DataFrame":
        return df.with_columns(
            pl.col("value").map_elements(lambda x: orjson.loads(x) if x else None, return_dtype=pl.Object),
            pl.col("schema").map_elements(lambda x: orjson.loads(x) if x else None, return_dtype=pl.Object),
        )

    if output_type == "DataFrame":
        # Convert strings to Pandas first, then parse to bypass PyArrow crashes
        df = df.to_pandas(use_pyarrow_extension_array=True)
        df["value"] = df["value"].apply(lambda x: orjson.loads(x) if pd.notna(x) else None)
        df["schema"] = df["schema"].apply(lambda x: orjson.loads(x) if pd.notna(x) else None)
        return df.set_index(["section", "key"])

    # 3. FASTEST APPROACH FOR DICT: Skip Pandas entirely
    # Convert Polars directly to a list of raw dictionaries and parse on the fly
    raw_records = df.to_dicts()
    new_output = {}

    for row in raw_records:
        sec = row["section"]
        key = row["key"]

        if sec not in new_output:
            new_output[sec] = {}

        new_output[sec][key] = {
            "id": row["id"],
            "value": orjson.loads(row["value"]) if row["value"] else None,
            "description": row["description"],
            "schema": orjson.loads(row["schema"]) if row["schema"] else None,
            "modified_date": row["modified_date"],
        }

    return new_output

insert(section, key, value, description, schema)

Inserts a new setting into the database

Parameters:

• section

  (str) –

  Section name

• key

  (str) –

  Setting name

• value

  (Any) –

  Value of the Setting

• description

  (str) –

  Description of the Setting

• schema

  (dict[str, Any]) –

  A JSON schema used to validate if the setting value is correct. If the value is not valid against the schema, an error will be raised.

Returns:

• int –

  The ID of the inserted setting

Source code in echo_postgres/settings.py

Python

@validate_call
def insert(
    self,
    section: str,
    key: str,
    value: Any,
    description: str,
    schema: dict[str, Any],
) -> int:
    """Inserts a new setting into the database

    Parameters
    ----------
    section : str
        Section name
    key : str
        Setting name
    value : Any
        Value of the Setting
    description : str
        Description of the Setting
    schema : dict[str, Any]
        A JSON schema used to validate if the setting value is correct. If the value is not valid against the schema, an error will be raised.

    Returns
    -------
    int
        The ID of the inserted setting
    """
    # validating the value against the schema
    try:
        jsonschema.validate(instance=value, schema=schema)
    except jsonschema.ValidationError as e:
        raise ValueError(f"Value {value} is not valid against the schema {schema}. Error: {e.message}") from e

    query = sql.SQL("""
        INSERT INTO performance.settings (section, key, value, description, schema)
        VALUES ({section}, {key}, {value}, {description}, {schema})
        RETURNING id
    """).format(
        section=sql.Literal(section),
        key=sql.Literal(key),
        value=sql.Literal(orjson.dumps(value).decode("utf-8")),
        description=sql.Literal(description),
        schema=sql.Literal(orjson.dumps(schema).decode("utf-8")),
    )

    return self._perfdb.conn.execute(query).fetchone()[0]

update(setting_id, section=None, key=None, value=None, description=None, schema=None)

Updates a setting in the database. At least one field must be provided for update.

Parameters:

• setting_id

  (int) –

  ID of the setting to update

• section

  (str, default: None ) –

  Section name, by default None (will not be updated)

• key

  (str, default: None ) –

  Setting name, by default None (will not be updated)

• value

  (Any, default: None ) –

  Value of the Setting, by default None (will not be updated)

• description

  (str, default: None ) –

  Description of the Setting, by default None (will not be updated)

• schema

  (dict[str, Any], default: None ) –

  A JSON schema used to validate if the setting value is correct. If the value is not valid against the schema, an error will be raised. By default None (will not be updated)

Returns:

• None –

Source code in echo_postgres/settings.py

Python

@validate_call
def update(
    self,
    setting_id: int,
    section: str | None = None,
    key: str | None = None,
    value: Any | None = None,
    description: str | None = None,
    schema: dict[str, Any] | None = None,
) -> None:
    """Updates a setting in the database. At least one field must be provided for update.

    Parameters
    ----------
    setting_id : int
        ID of the setting to update
    section : str, optional
        Section name, by default None (will not be updated)
    key : str, optional
        Setting name, by default None (will not be updated)
    value : Any, optional
        Value of the Setting, by default None (will not be updated)
    description : str, optional
        Description of the Setting, by default None (will not be updated)
    schema : dict[str, Any], optional
        A JSON schema used to validate if the setting value is correct. If the value is not valid against the schema, an error will be raised. By default None (will not be updated)

    Returns
    -------
    None
    """
    # checking if at least one field is provided for update
    if section is None and key is None and value is None and description is None and schema is None:
        raise ValueError("At least one field must be provided for update")

    # validating the value against the schema if value is provided
    if value is not None and schema is not None:
        try:
            jsonschema.validate(instance=value, schema=schema)
        except jsonschema.ValidationError as e:
            raise ValueError(f"Value {value} is not valid against the schema {schema}. Error: {e.message}") from e

    set_clauses = []
    if section is not None:
        set_clauses.append(sql.SQL("section = {section}").format(section=sql.Literal(section)))
    if key is not None:
        set_clauses.append(sql.SQL("key = {key}").format(key=sql.Literal(key)))
    if value is not None:
        set_clauses.append(sql.SQL("value = {value}").format(value=sql.Literal(orjson.dumps(value).decode("utf-8"))))
    if description is not None:
        set_clauses.append(sql.SQL("description = {description}").format(description=sql.Literal(description)))
    if schema is not None:
        set_clauses.append(sql.SQL("schema = {schema}").format(schema=sql.Literal(orjson.dumps(schema).decode("utf-8"))))
    set_clause = sql.SQL(", ").join(set_clauses)

    query = sql.SQL("""
        UPDATE performance.settings
        SET {set_clause}
        WHERE id = {id}
    """).format(
        set_clause=set_clause,
        id=sql.Literal(setting_id),
    )

    self._perfdb.conn.execute(query)