Build consult requests
Selection requests
Build all consult commands.
build_group_join(command, sqle, built_commands)
Build a group join request.
Parameters: |
|
---|
Returns: |
|
---|
Exceptions: |
|
---|
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: |
|
---|
Returns: |
|
---|
Exceptions: |
|
---|
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: |
|
---|
Returns: |
|
---|
Exceptions: |
|
---|
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: |
|
---|
Returns: |
|
---|
Exceptions: |
|
---|
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: |
|
---|
Returns: |
|
---|
Exceptions: |
|
---|
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: |
|
---|
Returns: |
|
---|
Exceptions: |
|
---|
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: |
|
---|
Returns: |
|
---|
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: |
|
---|
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: |
|
---|
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: |
|
---|
Returns: |
|
---|
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: |
|
---|
Returns: |
|
---|
Exceptions: |
|
---|
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: |
|
---|
Returns: |
|
---|
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: |
|
---|
Returns: |
|
---|
Exceptions: |
|
---|
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: |
|
---|
Returns: |
|
---|
Exceptions: |
|
---|
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: |
|
---|
Returns: |
|
---|
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)