Other functions

Functions used in py-linq-sql.

get_columns_name(mdps)

Get all column name.

Parameters:
  • mdps (Union[py_linq_sql.utils.classes.magicdotpath.MagicDotPath, List[py_linq_sql.utils.classes.magicdotpath.MagicDotPath], Tuple[py_linq_sql.utils.classes.magicdotpath.MagicDotPath], Dict[str, py_linq_sql.utils.classes.magicdotpath.MagicDotPath]]) – MagicDotPath for which we want the column name.

Returns:
  • List[str] – All of paths and columns.

Exceptions:
  • TypeOperatorError – Indirect raise by MagicDotPathWithOp.col_name.

Source code in py_linq_sql/utils/functions/other_functions.py
def get_columns_name(
    mdps: MagicDotPath
    | List[MagicDotPath]
    | Tuple[MagicDotPath]
    | Dict[str, MagicDotPath],
) -> List[str]:
    """
    Get all column name.

    Args:
        mdps: MagicDotPath for which we want the column name.

    Returns:
        All of paths and columns.

    Raises:
        TypeOperatorError: Indirect raise by `MagicDotPathWithOp.col_name`.
    """
    result = []

    match mdps:
        case BaseMagicDotPath():
            col_name = mdps.col_name()
            result.append(
                f"_{_short_columns_default_name(col_name)}"
                if len(col_name) > 58
                else f"{col_name}",
            )
        case tuple() | list():
            for element in mdps:
                col_name = element.col_name()
                result.append(
                    f"_{_short_columns_default_name(col_name)}"
                    if len(col_name) > 58
                    else f"{col_name}",
                )
        case dict():
            for key in mdps:
                result.append(_col_name_validator(key))

    return _fix_same_column_name(result)

get_good_type(cast_type)

Get the good type as str from a cast_type.

Parameters:
  • cast_type (type) – Type in which we want to cast the path.

Returns:
  • str | None – SQL type with '))' to cast in sql command. None if we give the wrong type.

Source code in py_linq_sql/utils/functions/other_functions.py
def get_good_type(cast_type: type) -> str | None:
    """
    Get the good type as str from a cast_type.

    Args:
        cast_type: Type in which we want to cast the path.

    Returns:
        SQL type with '))' to cast in sql command. None if we give the wrong type.
    """
    if cast_type in (int, float, Decimal):
        return "decimal))"
    if cast_type == date:
        return "date))"
    return None

get_json(data)

Get a json from data.

Parameters:
  • data (Dict[Any, Any]) – data we want to have in json.

Returns:
  • str – A json contains data.

Exceptions:
  • ValueError – Indirect raise by json.dumps.

Source code in py_linq_sql/utils/functions/other_functions.py
def get_json(data: Dict[Any, Any]) -> str:
    """
    Get a json from data.

    Args:
        data: data we want to have in json.

    Returns:
        A json contains data.

    Raises:
        ValueError: Indirect raise by `json.dumps`.
    """
    return json.dumps(data)

pretty_print(record)

Print a record in a pretty table with rich.

Parameters:
  • record (Enumerable) – Record to display.

Exceptions:
  • rich.errors.ConsoleError – Indirect raise by rich.table

  • rich.console.CaptureError – Indirect raise by rich.console

Source code in py_linq_sql/utils/functions/other_functions.py
def pretty_print(record: Enumerable) -> None:
    """
    Print a record in a pretty table with rich.

    Args:
        record: Record to display.

    Raises:
        rich.errors.ConsoleError: Indirect raise by `rich.table`
        rich.console.CaptureError: Indirect raise by `rich.console`
    """
    if record is None:
        Console().print("Record is None.")
    else:
        rec_in_list = record.to_list()
        if not rec_in_list:
            Console().print("Empty Enumerable.")
        else:
            table = Table(title="Record", show_lines=True)

            for name in rec_in_list[0]._fields:
                table.add_column(name, justify="center", no_wrap=True)

            for element in rec_in_list:
                table.add_row(*map(str, element))

            Console().print(table)
            count = len(rec_in_list)
            row_repr = f"({count} rows)" if count > 1 else f"({count} row)"
            Console().print(row_repr)

safe(connection, name)

Secure a column or a table for a request.

Parameters:
  • name (str) – Name of the column or table we want to secure.

Returns:
  • str – Name but verified by psycopg.sql.Identifier.

Exceptions:
  • psycopg.Error – Indirect raise by sql.Identifier or as_string.

Source code in py_linq_sql/utils/functions/other_functions.py
def safe(connection: Connection, name: str) -> str:
    """
    Secure a column or a table for a request.

    Args:
        name: Name of the column or table we want to secure.

    Returns:
        Name but verified by psycopg.sql.Identifier.

    Raises:
        psycopg.Error: Indirect raise by `sql.Identifier` or `as_string`.
    """
    return sql.Identifier(name).as_string(connection)