Build API
Build requests
Build request for execution.
dispatch_build(command, sqle, built_commands)
Dispatch command depending of his command type.
Parameters: |
|
---|
Returns: |
|
---|
Exceptions: |
|
---|
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: |
|
---|
Returns: |
|
---|
Exceptions: |
|
---|
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: |
|
---|
Returns: |
|
---|
Exceptions: |
|
---|
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: |
|
---|
Returns: |
|
---|
Exceptions: |
|
---|
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: |
|
---|
Returns: |
|
---|
Exceptions: |
|
---|
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: |
|
---|
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)
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: |
|
---|
Returns: |
|
---|
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: |
|
---|
Returns: |
|
---|
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: |
|
---|
Returns: |
|
---|
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: |
|
---|
Returns: |
|
---|
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: |
|
---|
Returns: |
|
---|
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: |
|
---|
Returns: |
|
---|
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: |
|
---|
Returns: |
|
---|
Exceptions: |
|
---|
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: |
|
---|
Returns: |
|
---|
Exceptions: |
|
---|
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: |
|
---|
Returns Request to execute.
Exceptions: |
|
---|
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))