Build alter requests

Build all alter commands.

build_delete(command, sqle, built_commands)

Build a delete request.

Parameters:
  • command (Command) – Command to build.

  • sqle (SQLEnumerableData) – SQLEnumerable with connection, flags, list of commands and a table.

  • built_commands (Set[int]) – All commands that have already been built.

Returns:
  • str – Request to execute.

Exceptions:
  • DeleteError – If len(sqle.cmd) > 1 and command.args.armageddon).

  • NeedWhereError – If len(sqle.cmd) < 2.

  • psycopg.Error – Indirect raise by build_where.

  • TableError – Indirect raise by build_where.

  • TypeError – Indirect raise by build_where.

  • TypeOperatorError – Indirect raise by build_where.

Source code in py_linq_sql/build_request/alter.py
def build_delete(
    command: Command,
    sqle: SQLEnumerableData,
    built_commands: Set[int],
) -> str:
    """
    Build a delete request.

    Args:
        command: Command to build.
        sqle: SQLEnumerable with connection, flags, list of commands and a table.
        built_commands: All commands that have already been built.

    Returns:
        Request to execute.

    Raises:
        DeleteError: If len(sqle.cmd) > 1 and command.args.armageddon).
        NeedWhereError: If len(sqle.cmd) < 2.
        psycopg.Error: Indirect raise by `build_where`.
        TableError: Indirect raise by `build_where`.
        TypeError: Indirect raise by `build_where`.
        TypeOperatorError: Indirect raise by `build_where`.
    """
    armageddon = command.args.armageddon

    result = [f"DELETE FROM {sqle.table}"]

    if len(sqle.cmd) > 1 and armageddon:
        raise DeleteError("where")

    if armageddon:
        return result[0]

    if len(sqle.cmd) < 2:
        raise NeedWhereError()

    result.append(build_where(sqle, built_commands))

    # We use filter with None for the argument __function.
    # If we give None to the first element of filter
    # it will pass all the elements evaluate to false no matter why.
    #
    # We can have None in result if sqle.cmd contains commands
    # which will be evaluated later in build_where()
    return " ".join(filter(None, result))

build_insert(command, sqle)

Build an insert request for json table or relational table.

Parameters:
  • command (Command) – Command to build.

  • sqle (SQLEnumerableData) – SQLEnumerable with connection, flags, list of commands and a table.

Returns:
  • str – Request to execute.

Exceptions:
  • TypeError – Raise when the data had the wrong type or Indirect raise by _build_json_insert.

  • ValueError – Indirect raise by _build_values_for_insert_simple_type or _build_values_for_insert_list.

Source code in py_linq_sql/build_request/alter.py
def build_insert(command: Command, sqle: SQLEnumerableData) -> str:
    """
    Build an insert request for json table or relational table.

    Args:
        command: Command to build.
        sqle: SQLEnumerable with connection, flags, list of commands and a table.

    Returns:
        Request to execute.

    Raises:
        TypeError: Raise when the data had the wrong type or
            Indirect raise by `_build_json_insert`.
        ValueError: Indirect raise by `_build_values_for_insert_simple_type`
            or `_build_values_for_insert_list`.
    """
    column = command.args.column
    data = command.args.data

    result = [f"INSERT INTO {sqle.table}("]

    match column:
        case str():
            result.append(f"{column}")
            result.append(")")
        case _:
            result.append(", ".join(column))
            result.append(")")

    result.append("VALUES")

    match data:
        case tuple():
            result.append(_build_values_for_insert_tuple(data))
        case list():
            result.append(_build_values_for_insert_list(data))
        case _:
            result.append(_build_values_for_insert_simple_type(data))

    return " ".join(result)

build_update(command, sqle, built_commands)

Build an update request.

Parameters:
  • command (Command) – Command to build.

  • sqle (SQLEnumerableData) – SQLEnumerable with connection, flags, list of commands and a table.

  • built_commands (Set[int]) – All commands that have already been built.

Returns Request to execute.

Exceptions:
  • TooManyReturnValueError – If len of path > 1.

  • psycopg.Error – Indirect raise by build_where.

  • TableError – Indirect raise by build_where.

  • TypeError – Indirect raise by build_where.

  • TypeOperatorError – Indirect raise by build_where, BaseMagicDotPath._get_number_operator or BaseMagicDotPath._get_generic_operator.

Source code in py_linq_sql/build_request/alter.py
def build_update(
    command: Command,
    sqle: SQLEnumerableData,
    built_commands: Set[int],
) -> str:
    """
    Build an update request.

    Args:
        command: Command to build.
        sqle: SQLEnumerable with connection, flags, list of commands and a table.
        built_commands: All commands that have already been built.

    Returns
        Request to execute.

    Raises:
        TooManyReturnValueError: If len of path > 1.
        psycopg.Error: Indirect raise by `build_where`.
        TableError: Indirect raise by `build_where`.
        TypeError: Indirect raise by `build_where`.
        TypeOperatorError: Indirect raise by `build_where`,
            `BaseMagicDotPath._get_number_operator`
            or `BaseMagicDotPath._get_generic_operator`.
    """
    fquery = command.args.fquery  # pylint: disable=duplicate-code
    mdp_w_path = fquery(MagicDotPath(sqle.connection))
    path = get_path(mdp_w_path, sqle)

    if len(path) > 1:
        raise TooManyReturnValueError("Update")

    operand_1 = mdp_w_path.operand_1
    column = operand_1.column
    operand_2 = mdp_w_path.operand_2

    json = len(operand_1.attributes) > 1
    path_for_update = "-".join(operand_1.attributes[1:])

    result = [f"UPDATE {sqle.table} SET {column} ="]

    if json:
        result.append(
            f"""jsonb_set({column}, """
            f"""'{get_update_path(path_for_update)}'::text[], '""",
        )

    match operand_2:
        case str() if not json:
            result.append(f"'{operand_2}'")
        case str() if json:
            result.append(f'"{operand_2}"')
        case _:
            result.append(f"{operand_2}")

    if json:
        result.append("', false)")

    if len(sqle.cmd) > 1:
        result.append(build_where(sqle, built_commands))

    # We use filter with None for the argument __function.
    # If we give None to the first element of filter
    # it will pass all the elements evaluate to false no matter why.
    #
    # We can have None in result if sqle.cmd contains commands
    # which will be evaluated later in build_where()
    return " ".join(filter(None, result))