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