Build API

Build requests

Build request for execution.

dispatch_build(command, sqle, built_commands)

Dispatch command depending of his command type.

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 | None – Request to execute or None if command will run later.

Exceptions:
  • DeleteError – Indirect raise by _dispatch_alter.

  • LengthMismatchError – Indirect raise by build_join.

  • NeedWhereError – Indirect raise by _dispatch_alter.

  • psycopg.Error – Indirect raise by build_select, build_join, build_group_join, _dispatch_alter or _dispatch_one.

  • TableError – Indirect raise by build_select, build_join, build_group_join, _dispatch_alter or _dispatch_one.

  • TooManyReturnValueError – Indirect raise by _dispatch_alter.

  • TypeError – Indirect raise by build_select, build_join, build_group_join, _dispatch_alter, or _dispatch_one.

  • TypeOperatorError – Indirect raise by build_select, build_join, build_group_join, _dispatch_alter, or_dispatch_one.

  • UnknownCommandTypeError – If type of command not in Ct or indirect raise by build_select, build_join or build_group_join.

  • ValueError – Indirect raise by _dispatch_alter or _dispatch_one.

Source code in py_linq_sql/build_request/build.py
def dispatch_build(
    command: Command,
    sqle: SQLEnumerableData,
    built_commands: Set[int],
) -> str | None:
    """
    Dispatch command depending of his command type.

    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 or None if command will run later.

    Raises:
        DeleteError: Indirect raise by `_dispatch_alter`.
        LengthMismatchError: Indirect raise by `build_join`.
        NeedWhereError: Indirect raise by `_dispatch_alter`.
        psycopg.Error: Indirect raise by
            `build_select`, `build_join`, `build_group_join`, `_dispatch_alter`
            or `_dispatch_one`.
        TableError: Indirect raise by
            `build_select`, `build_join`, `build_group_join`, `_dispatch_alter`
            or `_dispatch_one`.
        TooManyReturnValueError: Indirect raise by `_dispatch_alter`.
        TypeError: Indirect raise by
            `build_select`, `build_join`, `build_group_join`, `_dispatch_alter`,
            or `_dispatch_one`.
        TypeOperatorError: Indirect raise by
            `build_select`, `build_join`, `build_group_join`, `_dispatch_alter`,
            or`_dispatch_one`.
        UnknownCommandTypeError: If type of command not in Ct or
            indirect raise by `build_select`, `build_join` or `build_group_join`.
        ValueError: Indirect raise by `_dispatch_alter` or `_dispatch_one`.
    """
    result = None

    match command.cmd_type:
        case Ct.SELECT:
            result = build_select(command, sqle, built_commands)
        case Ct.JOIN:
            result = build_join(command, sqle, built_commands)
        case Ct.GROUP_JOIN:
            result = build_group_join(command, sqle, built_commands)
        case command.cmd_type if command.cmd_type in [Ct.INSERT, Ct.UPDATE, Ct.DELETE]:
            result = _dispatch_alter(command, sqle, built_commands)
        case command.cmd_type if command.cmd_type in [Ct.ANY, Ct.ALL, Ct.CONTAINS]:
            result = _dispatch_one(command, sqle)
        case command.cmd_type if command.cmd_type in list(Ct):  # type: ignore[operator]
            # HACK: The ignore above it's just the time that mypy supports the Strenum.
            # https://github.com/python/mypy/issues
            pass
        # The following case is just an other security layers,
        # but we can't go in this case for the moment.
        case _:  # pragma: no cover
            # HACK: The time that mypy supports the Strenum.
            # https://github.com/python/mypy/issues
            command_cmd_type = cast(Ct, command.cmd_type)
            raise UnknownCommandTypeError(command_cmd_type.value)

    return result

build(sqle)

Build a list of commands from an SQLEnumerable.

Parameters:
  • sqle (SQLEnumerableData) – SQLEnumerable contains list of commands to build.

Returns:
  • str – Request to execute.

Exceptions:
  • DeleteError – Indirect raise by dispatch_build.

  • LengthMismatchError – Indirect raise by dispatch_build.

  • NeedWhereError – Indirect raise by dispatch_build.

  • psycopg.Error – Indirect raise by dispatch_build.

  • TableError – Indirect raise by dispatch_build.

  • TooManyReturnValueError – Indirect raise by dispatch_build.

  • TypeError – Indirect raise by dispatch_build.

  • TypeOperatorError – Indirect raise by dispatch_build.

  • UnknownCommandTypeError – Indirect raise by dispatch_build.

  • ValueError – Indirect raise by dispatch_build.

Source code in py_linq_sql/build_request/build.py
def build(sqle: SQLEnumerableData) -> str:
    """
    Build a list of commands from an SQLEnumerable.

    Args:
        sqle: SQLEnumerable contains list of commands to build.

    Returns:
        Request to execute.

    Raises:
        DeleteError: Indirect raise by `dispatch_build`.
        LengthMismatchError: Indirect raise by `dispatch_build`.
        NeedWhereError: Indirect raise by `dispatch_build`.
        psycopg.Error: Indirect raise by `dispatch_build`.
        TableError: Indirect raise by `dispatch_build`.
        TooManyReturnValueError: Indirect raise by `dispatch_build`.
        TypeError: Indirect raise by `dispatch_build`.
        TypeOperatorError: Indirect raise by `dispatch_build`.
        UnknownCommandTypeError: Indirect raise by `dispatch_build`.
        ValueError: Indirect raise by `dispatch_build`.
    """
    built_commands: Set[int] = set()
    commands = sqle.cmd
    if not commands:
        return ""

    result = []
    for idx, cmd in enumerate(commands):
        if idx not in built_commands:
            res = dispatch_build(cmd, sqle, built_commands)
            if res:
                result.append(res)
                built_commands.add(idx)

    # 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_select() or build_update()
    return " ".join(filter(None, result))

Build any, all, contains requests

Build all One commands.

build_all(command, sqle)

Build an all request.

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

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

Returns:
  • str – Request to execute.

Exceptions:
  • psycopg.Error – Indirect raise by get_predicates_as_str.

  • TableError – Indirect raise by get_predicates_as_str.

  • TypeError – Indirect raise by get_predicates_as_str.

  • TypeOperatorError – Indirect raise by get_predicates_as_str.

Source code in py_linq_sql/build_request/one.py
def build_all(command: Command, sqle: SQLEnumerableData) -> str:
    """
    Build an all request.

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

    Returns:
        Request to execute.

    Raises:
        psycopg.Error: Indirect raise by `get_predicates_as_str`.
        TableError: Indirect raise by `get_predicates_as_str`.
        TypeError: Indirect raise by `get_predicates_as_str`.
        TypeOperatorError: Indirect raise by `get_predicates_as_str`.
    """
    fquery = command.args.fquery

    result = [f"SELECT CASE WHEN ((SELECT COUNT(*) FROM {sqle.table} WHERE"]

    get_predicates_as_str(result, fquery, sqle)

    result.append(
        f") = (SELECT COUNT(*) FROM {sqle.table})) THEN 1 ELSE 0 END FROM {sqle.table}",
    )

    return " ".join(result)

build_any(command, sqle)

Build an any request.

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

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

Returns:
  • str – Request to execute.

Exceptions:
  • psycopg.Error – Indirect raise by get_predicates_as_str.

  • TableError – Indirect raise by get_predicates_as_str.

  • TypeError – Indirect raise by get_predicates_as_str.

  • TypeOperatorError – Indirect raise by get_predicates_as_str.

Source code in py_linq_sql/build_request/one.py
def build_any(command: Command, sqle: SQLEnumerableData) -> str:
    """
    Build an any request.

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

    Returns:
        Request to execute.

    Raises:
        psycopg.Error: Indirect raise by `get_predicates_as_str`.
        TableError: Indirect raise by `get_predicates_as_str`.
        TypeError: Indirect raise by `get_predicates_as_str`.
        TypeOperatorError: Indirect raise by `get_predicates_as_str`.
    """
    fquery = command.args.fquery

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

    if not fquery:
        return result[0]

    result.append("WHERE")

    get_predicates_as_str(result, fquery, sqle)

    return " ".join(result)

build_contains(command, sqle)

Build an contains request.

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

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

Returns:
  • str – Request to execute.

Exceptions:
  • psycopg.Error – Indirect raise by build_any.

  • TableError – Indirect raise by build_any.

  • TypeError – Indirect raise by build_any.

  • TypeOperatorError – Indirect raise by build_any.

  • ValueError – Indirect raise by _contains_dict.

Source code in py_linq_sql/build_request/one.py
def build_contains(command: Command, sqle: SQLEnumerableData) -> str:
    """
    Build an contains request.

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

    Returns:
        Request to execute.

    Raises:
        psycopg.Error: Indirect raise by `build_any`.
        TableError: Indirect raise by `build_any`.
        TypeError: Indirect raise by `build_any`.
        TypeOperatorError: Indirect raise by `build_any`.
        ValueError: Indirect raise by `_contains_dict`.
    """
    fquery = command.args.fquery

    match fquery:
        case dict():
            return _contains_dict(sqle, fquery)
        case _:
            return build_any(command, sqle)

Build consult requests

Selection requests

Build all consult commands.

build_group_join(command, sqle, built_commands)

Build a group join 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 – Sub request to execute.

Exceptions:
  • psycopg.Error – Indirect raise by _get_selected_by, get_path or _build_context_and_terminal.

  • TableError – Indirect raise by _get_selected_by, get_path or_build_context_and_terminal.

  • TypeError – Indirect raise by _get_selected_by, get_path or _build_context_and_terminal.

  • TypeOperatorError – Indirect raise by _get_selected_by, _build_context_and_terminal,BaseMagicDotPath._get_number_operatororBaseMagicDotPath._get_generic_operator`.

  • UnknownCommandTypeError – Indirect raise by _build_context_and_terminal.

Source code in py_linq_sql/build_request/consult.py
def build_group_join(  # pylint: disable=too-many-locals
    command: Command,
    sqle: SQLEnumerableData,
    built_commands: Set[int],
) -> str:
    """
    Build a group join 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:
        Sub request to execute.

    Raises:
        psycopg.Error: Indirect raise by
            `_get_selected_by`, `get_path` or `_build_context_and_terminal`.
        TableError: Indirect raise by
            `_get_selected_by`, `get_path` or`_build_context_and_terminal`.
        TypeError: Indirect raise by
            `_get_selected_by`, `get_path` or `_build_context_and_terminal`.
        TypeOperatorError: Indirect raise by
            `_get_selected_by`, _build_context_and_terminal`,
            `BaseMagicDotPath._get_number_operator`
            or `BaseMagicDotPath._get_generic_operator`.
        UnknownCommandTypeError: Indirect raise by `_build_context_and_terminal`.
    """
    result_function = command.args.result_function
    outer = command.args.outer
    inner = command.args.inner
    outer_key = command.args.outer_key
    inner_key = command.args.inner_key
    join_type = command.args.join_type

    obj_inner = MagicDotPath(inner.connection, with_table=inner.table)
    obj_outer = MagicDotPath(outer.connection, with_table=outer.table)

    mdps = result_function(obj_inner, obj_outer)
    selected, by = _get_selected_by(mdps)  # pylint: disable=invalid-name

    result = [f"SELECT {selected}"]

    result.append(f"FROM {outer.table} {join_type.as_str} JOIN {inner.table} ON")

    paths = DotMap(
        outer_key_paths=get_path(outer_key(obj_outer)),
        inner_key_paths=get_path(inner_key(obj_inner)),
    )

    _build_join_clauses(result, paths, join_type)
    _build_context_and_terminal(result, sqle, built_commands)

    result.append(f"GROUP BY {by}")

    return " ".join(result)

build_join(command, sqle, built_commands)

Build a join 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 – Sub request to execute.

Exceptions:
  • LengthMismatchError – If len of path the outer_key is not equal to len of path of the inner key.

  • psycopg.Error – Indirect raise by join_get_paths, get_path or _build_context_and_terminal.

  • TableError – Indirect raise by join_get_paths, get_path or _build_context_and_terminal.

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

  • TypeError – Indirect raise by join_get_paths, get_path or _build_context_and_terminal.

  • UnknownCommandTypeError – Indirect raise by _build_context_and_terminal.

Source code in py_linq_sql/build_request/consult.py
def build_join(  # pylint: disable=too-many-locals
    command: Command,
    sqle: SQLEnumerableData,
    built_commands: Set[int],
) -> str:
    """
    Build a join 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:
        Sub request to execute.

    Raises:
        LengthMismatchError: If len of path the outer_key is not equal to
            len of path of the inner key.
        psycopg.Error: Indirect raise by
            `join_get_paths`, `get_path` or `_build_context_and_terminal`.
        TableError: Indirect raise by
            `join_get_paths`, `get_path` or `_build_context_and_terminal`.
        TypeOperatorError: Indirect raise by
            `BaseMagicDotPath._get_number_operator`,
            `BaseMagicDotPath._get_generic_operator`, `get_columns_name`
            or `_build_context_and_terminal`.
        TypeError: Indirect raise by
            `join_get_paths`, `get_path` or `_build_context_and_terminal`.
        UnknownCommandTypeError: Indirect raise by `_build_context_and_terminal`.
    """
    outer = command.args.outer
    inner = command.args.inner
    outer_key = command.args.outer_key
    inner_key = command.args.inner_key
    result_function = command.args.result_function
    join_type = command.args.join_type

    paths = join_get_paths(outer, inner, inner_key, outer_key, result_function)

    if not len(paths.outer_key_paths) == len(paths.inner_key_paths):
        raise LengthMismatchError("outer_key_path", "inner_key_paths")

    result = ["SELECT"]

    if not paths.select_paths:
        result.append(f"{outer.table}.*, {inner.table}.*")
    else:
        obj_inner = MagicDotPath(inner.connection, with_table=inner.table)
        obj_outer = MagicDotPath(outer.connection, with_table=outer.table)
        mdp_select_paths = result_function(obj_inner, obj_outer)
        names = get_columns_name(mdp_select_paths)
        result.append(
            ", ".join(
                [f"{path} AS {name}" for path, name in zip(paths.select_paths, names)],
            ),
        )

    result.append(f"FROM {outer.table} {join_type.as_str} JOIN {inner.table} ON")

    _build_join_clauses(result, paths, join_type)
    _build_context_and_terminal(result, sqle, built_commands)

    return " ".join(filter(None, result))

build_select(command, sqle, built_commands)

Build a select 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:
  • TypeError – If all commands are not subclasses of BaseMagicDotPath or Indirect raise by _build_select_addon, get_one_predicate_as_str, _build_context_and_terminal or build_group_by.

  • psycopg.Error – Indirect raise by _build_select_addon, get_one_predicate_as_str, _build_context_and_terminal, or build_group_by.

  • TableError – Indirect raise by _build_select_addon, get_one_predicate_as_str, _build_context_and_terminal, or build_group_by.

  • TypeOperatorError – Indirect raise by _build_select_addon, BaseMagicDotPath._get_number_operator, BaseMagicDotPath._get_generic_operator, get_columns_name, _build_context_and_terminal or build_group_by.

  • UnknownCommandTypeError – Indirect raise by _build_select_addon or_build_context_and_terminal.

Source code in py_linq_sql/build_request/consult.py
def build_select(
    command: Command,
    sqle: SQLEnumerableData,
    built_commands: Set[int],
) -> str:
    """
    Build a select 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:
        TypeError: If all commands are not subclasses of BaseMagicDotPath
            or Indirect raise by `_build_select_addon`, `get_one_predicate_as_str`,
            `_build_context_and_terminal` or `build_group_by`.
        psycopg.Error: Indirect raise by `_build_select_addon`,
            `get_one_predicate_as_str`, `_build_context_and_terminal`,
            or `build_group_by`.
        TableError: Indirect raise by `_build_select_addon`,
            `get_one_predicate_as_str`, `_build_context_and_terminal`,
            or `build_group_by`.
        TypeOperatorError: Indirect raise by `_build_select_addon`,
            `BaseMagicDotPath._get_number_operator`,
            `BaseMagicDotPath._get_generic_operator`, `get_columns_name`,
            `_build_context_and_terminal` or `build_group_by`.
        UnknownCommandTypeError: Indirect raise by `_build_select_addon`
            or`_build_context_and_terminal`.
    """
    fquery = command.args.fquery  # pylint: disable=duplicate-code

    result = ["SELECT"]

    term = (
        _build_select_addon(sqle, built_commands)
        if sqle.flags.terminal
        in [Terminal.MAX, Terminal.MIN, Terminal.COUNT, Terminal.DISTINCT]
        else None
    )

    if term:
        result.append(term)

    if not term or sqle.flags.terminal in [Terminal.DISTINCT]:
        if not fquery:
            result.append("*")
        else:
            mdp_w_path = fquery(MagicDotPath(sqle.connection))
            match mdp_w_path:
                case BaseMagicDotPath():
                    paths = [get_one_predicate_as_str(sqle, mdp_w_path)]
                case tuple():
                    paths = [get_one_predicate_as_str(sqle, mdp) for mdp in mdp_w_path]
                case dict():
                    paths = [
                        get_one_predicate_as_str(sqle, mdp)
                        for mdp in mdp_w_path.values()
                    ]
                case _:
                    raise TypeError(
                        "You must put a MagicDotPath in lambda, see the documentation.",
                    )
            names = get_columns_name(mdp_w_path)
            result.append(
                ", ".join([f"{path} AS {name}" for path, name in zip(paths, names)]),
            )

    # If we have a Group_by we build this
    # (the last command because it's a terminal command)
    # we append all aggregate built by `build_group_by()`
    # and we add to the built_commands the index of this group_by.
    if sqle.flags.terminal == Terminal.GROUP_BY:
        aggregates = build_group_by(sqle.cmd[-1], sqle)
        result.append(f", {aggregates}")
        built_commands.add(len(sqle.cmd) - 1)

    if isinstance(sqle.table, str):
        result.append(f"FROM {sqle.table}")
    else:
        result.append(f"FROM ({sqle.table.get_command()}) AS subrequest")

    _build_context_and_terminal(result, sqle, built_commands)

    if sqle.flags.terminal == Terminal.GROUP_BY:
        result.append("GROUP BY")
        result.append(", ".join(list(path for path in paths)))

    return " ".join(filter(None, result))

Context requests

Build all context consult commands.

build_order_by(command, sqle, suffix='ASC')

Build an order_by/order_by_descending request.

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

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

  • suffix (str) – Suffix to add to define if we order by the begin or the end. By default: ASC (to begin).

Returns:
  • str – Sub request to execute.

Exceptions:
  • psycopg.Error – Indirect raise by get_paths.

  • TableError – Indirect raise by get_paths.

  • TypeError – Indirect raise by get_paths.

  • TypeOperatorError – Indirect raise by get_paths.

Source code in py_linq_sql/build_request/consult_context.py
def build_order_by(
    command: Command,
    sqle: SQLEnumerableData,
    suffix: str = "ASC",
) -> str:
    """
    Build an order_by/order_by_descending request.

    Args:
        command: Command to build.
        sqle: SQLEnumerable with connection, flags, list of commands and a table.
        suffix: Suffix to add to define if we order by the begin or the end.
            By default: ASC (to begin).

    Returns:
        Sub request to execute.

    Raises:
        psycopg.Error: Indirect raise by `get_paths`.
        TableError: Indirect raise by `get_paths`.
        TypeError: Indirect raise by `get_paths`.
        TypeOperatorError: Indirect raise by `get_paths`.
    """
    fquery = command.args.fquery
    paths = get_paths(fquery, sqle)

    result = ["ORDER BY"]

    result.append(", ".join([f"{path} {suffix}" for path in paths]))

    return " ".join(result)

build_where(sqle, built_commands)

Build all where request.

Parameters:
  • 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 – Sub request to execute.

Exceptions:
  • psycopg.Error – Indirect raise by _build_one_where.

  • TableError – Indirect raise by _build_one_where.

  • TypeError – Indirect raise by _build_one_where.

  • TypeOperatorError – Indirect raise by _build_one_where.

Source code in py_linq_sql/build_request/consult_context.py
def build_where(sqle: SQLEnumerableData, built_commands: Set[int]) -> str:
    """
    Build all where request.

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

    Returns:
        Sub request to execute.

    Raises:
        psycopg.Error: Indirect raise by `_build_one_where`.
        TableError: Indirect raise by `_build_one_where`.
        TypeError: Indirect raise by `_build_one_where`.
        TypeOperatorError: Indirect raise by `_build_one_where`.
    """
    commands = sqle.cmd
    first_where = False
    result = []

    for idx, cmd in enumerate(commands):
        if cmd.cmd_type == CommandType.WHERE:
            if not first_where:
                result.append(_build_one_where(cmd, sqle))
                first_where = True
                built_commands.add(idx)
            else:
                result.append(_build_one_where(cmd, sqle, first=False))
                built_commands.add(idx)

    return " ".join(result)

define_limit_offset(sqle, built_commands)

Define the final limit offset for an SQL command.

Parameters:
  • 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 – The final limit offset with the correct syntax.

Exceptions:
  • TypeError – If sqle.length is None.

Source code in py_linq_sql/build_request/consult_context.py
def define_limit_offset(sqle: SQLEnumerableData, built_commands: Set[int]) -> str:
    """
    Define the final limit offset for an SQL command.

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

    Returns:
        The final limit offset with the correct syntax.

    Raises:
        TypeError: If sqle.length is None.
    """
    commands = sqle.cmd

    begin, end = 0, sqle.length

    # We ignore type on end because if we enter in this function,
    # we are sure that the size has been calculated beforehand.

    for idx, cmd in enumerate(commands):
        match cmd.cmd_type:
            case CommandType.TAKE:
                begin, end = define_one_begin_end(
                    begin,
                    cast(int, end),
                    cmd.args.number,
                    CommandType.TAKE,
                )
                built_commands.add(idx)
            case CommandType.SKIP:
                begin, end = define_one_begin_end(
                    begin,
                    cast(int, end),
                    cmd.args.number,
                    CommandType.SKIP,
                )
                built_commands.add(idx)
            case CommandType.TAKE_LAST:
                begin, end = define_one_begin_end(
                    begin,
                    cast(int, end),
                    cmd.args.number,
                    CommandType.TAKE_LAST,
                )
                built_commands.add(idx)
            case CommandType.SKIP_LAST:
                begin, end = define_one_begin_end(
                    begin,
                    cast(int, end),
                    cmd.args.number,
                    CommandType.SKIP_LAST,
                )
                built_commands.add(idx)
            case _:
                pass

    limit = cast(int, end) - begin
    offset = begin
    return f"LIMIT {limit} OFFSET {offset}"

define_one_begin_end(begin, end, number, type_)

Define the begin, end depending on the type of command and the old begin, end.

Parameters:
  • begin (int) – The old begin, the first time its 0.

  • end (int) – The old end, the first time its the length of the table with all conditions.

  • number (int) – The number use to calculate the new begin or end.

  • type_ (py_linq_sql.utils.classes.enum.CommandType | str) – Type of the command who was called.

Returns:
  • tuple[int, int] – The new begin and the new end.

Examples:

With take:

>>> define_one_begin_end(0, 7, 3, CommandType.TAKE)
(0, 3)

With skip:

>>> define_one_begin_end(0, 7, 3, CommandType.SKIP)
(3, 7)

With take_last:

>>> define_one_begin_end(0, 7, 3, CommandType.TAKE_LAST)
(4, 7)

With skip_last:

>>> define_one_begin_end(0, 7, 3, CommandType.SKIP_LAST)
(0, 4)
Source code in py_linq_sql/build_request/consult_context.py
def define_one_begin_end(
    begin: int,
    end: int,
    number: int,
    type_: CommandTypeOrStr,
) -> tuple[int, int]:
    """
    Define the begin, end depending on the type of command and the old begin, end.

    Args:
        begin: The old begin, the first time its 0.
        end: The old end, the first time its the length of the table
            with all conditions.
        number: The number use to calculate the new begin or end.
        type_: Type of the command who was called.

    Returns:
        The new begin and the new end.

    Examples:
        With take:
        >>> define_one_begin_end(0, 7, 3, CommandType.TAKE)
        (0, 3)

        With skip:
        >>> define_one_begin_end(0, 7, 3, CommandType.SKIP)
        (3, 7)

        With take_last:
        >>> define_one_begin_end(0, 7, 3, CommandType.TAKE_LAST)
        (4, 7)

        With skip_last:
        >>> define_one_begin_end(0, 7, 3, CommandType.SKIP_LAST)
        (0, 4)
    """
    match type_:
        case CommandType.TAKE:
            end = min(begin + number, end)
        case CommandType.SKIP:
            begin = begin + number
        case CommandType.TAKE_LAST:
            new_begin = end - number
            if new_begin >= begin:
                begin = new_begin
        case CommandType.SKIP_LAST:
            end = end - number

    if end - begin <= 0:
        raise ReturnEmptyEnumerable
    return begin, end

Terminal requests

Build all terminal consult commands.

build_count()

Build a count request.

Returns:
  • str – Sub request to execute.

Source code in py_linq_sql/build_request/consult_terminal.py
def build_count() -> str:
    """
    Build a count request.

    Returns:
        Sub request to execute.
    """
    return "COUNT(*)"

build_distinct()

Build a distinct request.

Returns:
  • str – Sub request to execute.

Source code in py_linq_sql/build_request/consult_terminal.py
def build_distinct() -> str:
    """
    Build a distinct request.

    Returns:
        Sub request to execute.
    """
    return "DISTINCT"

build_except(command)

Build an except request.

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

Returns:
  • str – Sub request to execute.

Source code in py_linq_sql/build_request/consult_terminal.py
def build_except(command: Command) -> str:
    """
    Build an except request.

    Args:
        command: Command to build.

    Returns:
        Sub request to execute.
    """
    return f"EXCEPT {command.args.exclude_cmd}"

build_group_by(command, sqle)

Build a group_by request.

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

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

Returns:
  • str – Sub request to execute.

Exceptions:
  • psycopg.Error – Indirect raise by get_aggregate.

  • TableError – Indirect raise by get_aggregate.

  • TypeError – Indirect raise by get_aggregate.

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

Source code in py_linq_sql/build_request/consult_terminal.py
def build_group_by(
    command: Command,
    sqle: SQLEnumerableData,
) -> str:
    """
    Build a group_by request.

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

    Returns:
        Sub request to execute.

    Raises:
        psycopg.Error: Indirect raise by `get_aggregate`.
        TableError: Indirect raise by `get_aggregate`.
        TypeError: Indirect raise by `get_aggregate`.
        TypeOperatorError: Indirect raise by `BaseMagicDotPath._get_number_operator`
            or `BaseMagicDotPath._get_generic_operator`.
    """
    aggreg_fquery = command.args.aggreg_fquery
    connection = sqle.connection

    mdp = MagicDotPath(connection)
    mdp_aggregate = aggreg_fquery(mdp)

    aggregate = get_aggregate(mdp_aggregate, sqle)

    return aggregate

build_intersect(command)

Build an intersect request.

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

Returns:
  • str – Sub request to execute.

Source code in py_linq_sql/build_request/consult_terminal.py
def build_intersect(command: Command) -> str:
    """
    Build an intersect request.

    Args:
        command: Command to build.

    Returns:
        Sub request to execute.
    """
    built_sqle_2 = command.args.built_sqle_2

    result = ["INTERSECT"]

    result.append(built_sqle_2)

    return " ".join(result)

build_max(command, sqle)

Build a max request.

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

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

Returns:
  • str – Sub request to execute.

Exceptions:
  • TypeError – If not get_good_type(cast_type) or indirect raise by get_paths.

  • psycopg.Error – Indirect raise by get_paths.

  • TableError – Indirect raise by get_paths.

  • TypeOperatorError – Indirect raise by get_paths.

Source code in py_linq_sql/build_request/consult_terminal.py
def build_max(command: Command, sqle: SQLEnumerableData) -> str:
    """
    Build a max request.

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

    Returns:
        Sub request to execute.

    Raises:
        TypeError: If not get_good_type(cast_type) or indirect raise by `get_paths`.
        psycopg.Error: Indirect raise by `get_paths`.
        TableError: Indirect raise by `get_paths`.
        TypeOperatorError: Indirect raise by `get_paths`.
    """
    fquery = command.args.fquery
    cast_type = command.args.cast_type
    path = get_paths(fquery, sqle)[0]

    if not cast_type or cast_type == str:
        path = get_paths(fquery, sqle, True)[0]
        return f"MAX({path})"

    path = get_paths(fquery, sqle)[0]
    result = [f"MAX(CAST({path} AS"]

    casted_type = get_good_type(cast_type)

    if not casted_type:
        raise TypeError(f"Max take only int, float or date type, not {cast_type}")

    result.append(casted_type)

    return " ".join(result)

build_min(command, sqle)

Build a min request.

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

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

Returns:
  • str – Sub request to execute.

Exceptions:
  • TypeError – If not get_good_type(cast_type) or indirect raise by get_paths.

  • psycopg.Error – Indirect raise by get_paths.

  • TableError – Indirect raise by get_paths.

  • TypeOperatorError – Indirect raise by get_paths.

Source code in py_linq_sql/build_request/consult_terminal.py
def build_min(command: Command, sqle: SQLEnumerableData) -> str:
    """
    Build a min request.

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

    Returns:
        Sub request to execute.

    Raises:
        TypeError: If not get_good_type(cast_type) or indirect raise by `get_paths`.
        psycopg.Error: Indirect raise by `get_paths`.
        TableError: Indirect raise by `get_paths`.
        TypeOperatorError: Indirect raise by `get_paths`.
    """
    fquery = command.args.fquery
    cast_type = command.args.cast_type

    if not cast_type or cast_type == str:
        path = get_paths(fquery, sqle, True)[0]
        return f"MIN({path})"

    path = get_paths(fquery, sqle)[0]
    result = [f"MIN(CAST({path} AS"]

    casted_type = get_good_type(cast_type)

    if not casted_type:
        raise TypeError(f"Min take only int, float or date type, not {cast_type}")

    result.append(casted_type)

    return " ".join(result)

build_union(command)

Build an union request.

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

Returns:
  • str – Sub request to execute.

Source code in py_linq_sql/build_request/consult_terminal.py
def build_union(command: Command) -> str:
    """
    Build an union request.

    Args:
        command: Command to build.

    Returns:
        Sub request to execute.
    """
    built_sqle_2 = command.args.built_sqle_2
    all_ = command.args.all_

    result = ["UNION ALL"] if all_ else ["UNION"]

    result.append(built_sqle_2)

    return " ".join(result)

Build aggregate for group by requests

Functions for aggregation in group_by.

avg(mdp, cast_type=<class 'float'>)

Aggregate function to make a AVG.

Parameters:
  • mdp (MagicDotPath) – A MagicDotPath to give the path of the function.

  • cast_type (type) – Type in which we want to cast the path(s). Its optional. By default: float

Returns:
  • MagicDotPathAggregate – MagicDotPathAggregate with the mdp and the type of aggregation.

Source code in py_linq_sql/build_request/consult_aggregate.py
def avg(mdp: MagicDotPath, cast_type: type = float) -> MagicDotPathAggregate:
    """
    Aggregate function to make a AVG.

    Args:
        mdp: A MagicDotPath to give the path of the function.
        cast_type: Type in which we want to cast the path(s). Its optional.
            By default: float

    Returns:
        MagicDotPathAggregate with the mdp and the type of aggregation.
    """
    return MagicDotPathAggregate(mdp, AggregateType.AVG, cast_type)

concat(mdp, separator)

Aggregate function to make a CONCAT.

Parameters:
  • mdp (MagicDotPath) – A MagicDotPath to give the path of the function.

Returns:
  • MagicDotPathAggregate – MagicDotPathAggregate with the mdp, the type of aggregation and the separator.

Source code in py_linq_sql/build_request/consult_aggregate.py
def concat(
    mdp: MagicDotPath,
    separator: str,
) -> MagicDotPathAggregate:
    """
    Aggregate function to make a CONCAT.

    Args:
        mdp: A MagicDotPath to give the path of the function.

    Returns:
        MagicDotPathAggregate with the mdp, the type of aggregation and the separator.
    """
    return MagicDotPathAggregate(
        mdp,
        AggregateType.CONCAT,
        separator=separator,
        cast_type=str,
    )

count(mdp, cast_type=<class 'float'>)

Aggregate function to make a COUNT.

Parameters:
  • - (mdp) – A MagicDotPath to give the path of the function.

  • - (cast_type) – Type in which we want to cast the path(s). Its optional. By default: float

Returns:
  • MagicDotPathAggregate – MagicDotPathAggregate with the mdp and the type of aggregation.

Source code in py_linq_sql/build_request/consult_aggregate.py
def count(mdp: MagicDotPath, cast_type: type = float) -> MagicDotPathAggregate:
    """
    Aggregate function to make a COUNT.

    Args:
        - mdp: A MagicDotPath to give the path of the function.
        - cast_type: Type in which we want to cast the path(s). Its optional.
            By default: float

    Returns:
        MagicDotPathAggregate with the mdp and the type of aggregation.
    """
    return MagicDotPathAggregate(mdp, AggregateType.COUNT, cast_type)

max(mdp, cast_type=<class 'float'>)

Aggregate function to make a MAX.

Parameters:
  • mdp (MagicDotPath) – A MagicDotPath to give the path of the function.

  • cast_type (type) – Type in which we want to cast the path(s). Its optional. By default: float

Returns:
  • MagicDotPathAggregate – MagicDotPathAggregate with the mdp and the type of aggregation.

Source code in py_linq_sql/build_request/consult_aggregate.py
def max(  # noqa: A001
    mdp: MagicDotPath,
    cast_type: type = float,
) -> MagicDotPathAggregate:
    """
    Aggregate function to make a MAX.

    Args:
        mdp: A MagicDotPath to give the path of the function.
        cast_type: Type in which we want to cast the path(s). Its optional.
            By default: float

    Returns:
        MagicDotPathAggregate with the mdp and the type of aggregation.
    """
    return MagicDotPathAggregate(mdp, AggregateType.MAX, cast_type)

min(mdp, cast_type=<class 'float'>)

Aggregate function to make a MIN.

Parameters:
  • mdp (MagicDotPath) – A MagicDotPath to give the path of the function.

  • cast_type (type) – Type in which we want to cast the path(s). Its optional. By default: float

Returns:
  • MagicDotPathAggregate – MagicDotPathAggregate with the mdp and the type of aggregation.

Source code in py_linq_sql/build_request/consult_aggregate.py
def min(  # noqa: A001
    mdp: MagicDotPath,
    cast_type: type = float,
) -> MagicDotPathAggregate:
    """
    Aggregate function to make a MIN.

    Args:
        mdp: A MagicDotPath to give the path of the function.
        cast_type: Type in which we want to cast the path(s). Its optional.
            By default: float

    Returns:
        MagicDotPathAggregate with the mdp and the type of aggregation.
    """
    return MagicDotPathAggregate(mdp, AggregateType.MIN, cast_type)

sum(mdp, cast_type=<class 'float'>)

Aggregate function to make a SUM.

Parameters:
  • mdp (MagicDotPath) – A MagicDotPath to give the path of the function.

  • cast_type (type) – Type in which we want to cast the path(s). Its optional. By default: float

Returns:
  • MagicDotPathAggregate – MagicDotPathAggregate with the mdp and the type of aggregation.

Source code in py_linq_sql/build_request/consult_aggregate.py
def sum(  # noqa: A001
    mdp: MagicDotPath,
    cast_type: type = float,
) -> MagicDotPathAggregate:
    """
    Aggregate function to make a SUM.

    Args:
        mdp: A MagicDotPath to give the path of the function.
        cast_type: Type in which we want to cast the path(s). Its optional.
            By default: float

    Returns:
        MagicDotPathAggregate with the mdp and the type of aggregation.
    """
    return MagicDotPathAggregate(mdp, AggregateType.SUM, cast_type)

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))