API documentation
SQLEnumerable API
SQLEnumerable objects and all methods of this class generate a SQL jsonb request.
SQLEnumerable
dataclass
SQLEnumerable objects with some specific methods.
Attributes:
Name | Type | Description |
---|---|---|
connection |
Connection |
Connection on which we want to execute the request. |
table |
str | SQLEnumerable |
Table or an other request on which we want to execute the request. |
cmd |
List[Command] |
Commands we want to execute. |
flags |
Flags |
All flags use to know the statement of the request. |
executed |
bool |
Executed status, True if is already executed False otherwise. |
length |
int | None |
Length of the result of the request if we need else None. |
Source code in py_linq_sql/sql_enumerable/sql_enumerable.py
@dataclass
class SQLEnumerable:
"""
SQLEnumerable objects with some specific methods.
Attributes:
connection (Connection): Connection on which we want to execute the request.
table (str | SQLEnumerable): Table or an other request on which
we want to execute the request.
cmd (List[Command]): Commands we want to execute.
flags (Flags): All flags use to know the statement of the request.
executed (bool): Executed status, True if is already executed False otherwise.
length (int | None): Length of the result of the request if we need else None.
"""
connection: Connection
table: str | SQLEnumerable
cmd: List[Command] = field(default_factory=list, init=False)
flags: Flags = field(default_factory=Flags)
executed: bool = False
length: int | None = None
def __post_init__(self) -> None:
"""
Initialize the command list and safe the table if it's an str.
Raises:
psycopg.Error: Indirect raise by `safe`.
TablePermissionDeniedError: Indirect raise by
`is_valid_table_name_with_white_and_black_list`.
"""
if isinstance(self.table, str):
tables = self.table.split(".") if "." in self.table else [self.table]
safed_tables = []
for table in tables:
if not is_valid_table_name_with_white_and_black_list(table):
raise TablePermissionDeniedError(table)
safed_tables.append(safe(self.connection, table))
self.table = ".".join([f'"{stab[1:-1]}"' for stab in safed_tables])
self.cmd = []
def __eq__(self, other: Any) -> bool:
"""
Try equality between two SQL_Enumerable.
SQL_Enumerable_1 == SQL_Enumerable_2.
"""
return equality(self, other)
def __ne__(self, other: Any) -> bool:
"""
Try no-equality between two SQL_Enumerable.
SQL_Enumerable_1 != SQL_Enumerable_2.
"""
return bool(not self.__eq__(other))
def copy(self) -> SQLEnumerable:
"""Create a shallow copy of self."""
return SQLEnumerable(
self.connection,
self._format_table(),
self.flags.copy(),
self.executed,
self.length,
)
def _format_table(self) -> str | SQLEnumerable:
if isinstance(self.table, str):
return self.table[1:-1]
return self.table
def _only_where_command(self) -> bool:
"""Check if all methods call before are where commands."""
for command in self.cmd:
if not command.cmd_type == Ct.WHERE:
return False
return True
def _check_select(
self,
sqle: SQLEnumerable,
caller_name: str,
) -> None:
"""
Check if all select are correct.
Check if we we have select in self and sqle.
Check if the length of those select are equals.
To understand the 2 first expressions you can read:
https://stackoverflow.com/questions/9542738/python-find-in-list/9542768#9542768
With those expressions we get the first element who validate
`command.cmd_type == Ct.SELECT` if no element validates we get None.
Args:
sqle: The second sqle to check (the first is self).
caller_name: Name of the function call _check_select
Raises:
NeedSelectError: If there is no select in self, or sqle (or both).
LengthMismatchError: If the number of column selected are not equal.
TypeError: Indirect raise by `get_path`.
psycopg.Error: Indirect raise by `get_path`.
TableError: Indirect raise by `get_path`.
"""
self_select = next(
(command for command in self.cmd if command.cmd_type == Ct.SELECT),
None,
)
sqle_select = next(
(command for command in sqle.cmd if command.cmd_type == Ct.SELECT),
None,
)
match (self_select, sqle_select):
case (None, _) | (_, None) | (None, None):
raise NeedSelectError(caller_name)
# From now on we recover fquerys and MagicDotPath to do test on it.
self_fquery = self_select.args.fquery # type: ignore[union-attr]
sqle_fquery = sqle_select.args.fquery # type: ignore[union-attr]
mdp_self = self_fquery(MagicDotPath(self.connection)) if self_fquery else "all"
mdp_sqle = sqle_fquery(MagicDotPath(sqle.connection)) if sqle_fquery else "all"
match (mdp_self, mdp_sqle):
case (str(), str()):
return
case (_, str()) | (str(), _):
raise LengthMismatchError("select of self", "select of sqle")
size_select_self = len(get_path(mdp_self))
size_select_sqle = len(get_path(mdp_sqle))
if size_select_self != size_select_sqle:
raise LengthMismatchError("select of self", "select of sqle")
def _check_legality_terminal_alter(self, cmd: Ctos) -> None:
"""
Check if we can make a terminal or alter command compared to previous commands.
Args:
cmd: Command that we want to add to the command list.
Raises:
AlterError: If the command list already contains an alter command.
self.flags.alter == True.
TerminalError: If the command list already contains an terminal command.
self.flags.terminal == True.
"""
# HACK: The time that mypy supports the Strenum.
# https://github.com/python/mypy/issues
cmd = cast(Ct, cmd)
if self.flags.terminal:
raise TerminalError(cmd.value)
if self.flags.alter:
raise AlterError(cmd.value)
def _check_legality(self, cmd: Ctos) -> None:
"""
Check if we can make a command compared to previous commands.
Args:
cmd: Command that we want to add to the command list.
Raises:
AlterError: If the command list already contains an alter command.
self.flags.alter == True
or indirect raise by `_check_legality_terminal_alter`
AlreadyExecutedError: If this SQLEnumerable is already executed.
self.executed == True.
GroupByWithJoinError: If the command is a group by
and we already make a join.
cmd == Ct.GROUP_BY and self.flags.join == True.
MoreThanZeroError: If the command list isn't empty. len(self.cmd) > 0.
NeedSelectError: If we make an intersect
or union command with select before.
self.flags.select == False.
NoMaxOrMinAfterLimitOffsetError: If the command is type of max or min and
we make a command who uses limit and/or offset before.
cmd in [Ct.MAX, Ct.MIN] and self.flags.limit_offset == True.
OneError: If we make a one command before. self.flags.one == True.
OtherThanWhereError: If we make other than where command before.
self._only_where_command() == False.
SelectError: If we make a select before. self.flags.select == True.
TerminalError: If the command list already contains an terminal command.
self.flags.terminal == True
or indirect raise by `_check_legality_terminal_alter`.
UnknownCommandTypeError: If the command type is unknown. cmd not in Ct.
"""
if self.executed:
raise AlreadyExecutedError()
if self.flags.one:
raise OneError()
# HACK: All following HACK it's just the time that mypy supports the Strenum.
# https://github.com/python/mypy/issues
match cmd:
case Ct.SELECT:
if self.flags.select:
cmd = cast(Ct, cmd) # HACK
raise SelectError(cmd.value)
if self.flags.alter:
cmd = cast(Ct, cmd) # HACK
raise AlterError(cmd.value)
case (Ct.INSERT | Ct.ALL | Ct.ANY | Ct.CONTAINS):
if self.cmd:
cmd = cast(Ct, cmd) # HACK
raise MoreThanZeroError(cmd.value)
case (Ct.UPDATE | Ct.DELETE):
if not self._only_where_command():
raise OtherThanWhereError()
case Ct.WHERE:
if self.flags.terminal:
cmd = cast(Ct, cmd) # HACK
raise TerminalError(cmd.value)
case (Ct.MAX | Ct.MIN):
self._check_legality_terminal_alter(cmd)
if self.flags.limit_offset:
cmd = cast(Ct, cmd) # HACK
raise NoMaxOrMinAfterLimitOffsetError(cmd.value)
case (Ct.INTERSECT | Ct.UNION):
if not self.flags.select:
cmd = cast(Ct, cmd) # HACK
raise NeedSelectError(cmd.value)
self._check_legality_terminal_alter(cmd)
case Ct.GROUP_BY:
if self.flags.join:
raise GroupByWithJoinError()
self._check_legality_terminal_alter(cmd)
case cmd if cmd in list(Ct): # type: ignore[operator] # HACK
self._check_legality_terminal_alter(cmd)
# The following case is just an other security layers,
# but we can't go in this case for the moment.
case _: # pragma: no cover
cmd = cast(Ct, cmd) # HACK
raise UnknownCommandTypeError(cmd.value)
def _pre_build_for_count(
self,
include: List[Ctos],
suffix: Command,
) -> SQLEnumerableData:
"""
Build an SQLEnumerableData with the same attribute.
Get the correct commands list to count the length of the return.
Args:
include: Commands we want to keep.
suffix: Command added at the end.
Returns:
SQLEnumerableData with correct commands
"""
commands = self.cmd
pre_build_cmd = []
for cmd in commands:
if cmd.cmd_type in include:
pre_build_cmd.append(cmd)
if not pre_build_cmd:
pre_build_cmd.append(Command(Ct.SELECT, DotMap(fquery=None)))
pre_build_cmd.append(suffix)
hidden_sqle = SQLEnumerableData(
self.connection,
self.flags.copy(),
pre_build_cmd,
self.table,
None,
)
# limit_offset flag is false because we don't copy the limit_offset commands.
hidden_sqle.flags.limit_offset = False
# terminal flag is true because we add a count command (is a terminal command)
hidden_sqle.flags.terminal = Terminal.COUNT
return hidden_sqle
def _set_count(self) -> None:
"""
Set the count of a request's return in self.
Raises:
DeleteError: Indirect raise by `build`.
LengthMismatchError: Indirect raise by `build`.
NeedWhereError: Indirect raise by `build`.
psycopg.Error: Indirect raise by `build`.
TableError: Indirect raise by `build`.
TooManyReturnValueError: Indirect raise by `build`.
TypeError: Indirect raise by `build`.
TypeOperatorError: Indirect raise by `build`.
UnknownCommandTypeError: Indirect raise by `build`.
ValueError: Indirect raise by `build`.
"""
hidden_sqle = self._pre_build_for_count(
[Ct.SELECT, Ct.WHERE, Ct.EXCEPT_],
Command(Ct.COUNT, None),
)
cmd_to_execute = build(hidden_sqle)
cursor = get_cursor(self.connection)
cursor.execute(cmd_to_execute)
fetch = cursor.fetchone()
if fetch:
count = fetch.count
self.length = count
def get_command(self) -> str | None:
"""
Get the command as an str.
Returns:
The request as an str.
Raises:
DeleteError: Indirect raise by `build` or `_set_count`.
LengthMismatchError: Indirect raise by `build` or `_set_count`.
NeedWhereError: Indirect raise by `build` or `_set_count`.
psycopg.Error: Indirect raise by `build` or `_set_count`.
TableError: Indirect raise by `build` or `_set_count`.
TooManyReturnValueError: Indirect raise by `build` or `_set_count`.
TypeError: Indirect raise by `build` or `_set_count`.
TypeOperatorError: Indirect raise by `build` or `_set_count`.
UnknownCommandTypeError: Indirect raise by `build` or `_set_count`.
ValueError: Indirect raise by `build` or `_set_count`.
"""
if self.flags.limit_offset:
self._set_count()
hidden_sqle = SQLEnumerableData(
self.connection,
self.flags,
self.cmd,
self.table,
self.length,
)
try:
return build(hidden_sqle)
except ReturnEmptyEnumerable:
return None
# ------------------------
# | Alteration methods |
# ------------------------
def delete(
self,
fquery: LambdaMagicDotPath | None = None,
*,
armageddon: bool = False,
) -> SQLEnumerable:
"""
Add a DELETE command to the command list.
Args:
fquery: Lambda function to get the element to delete.
armageddon: True if you want to delete ALL your table, by default is False.
Returns:
Self.
Raises:
DeleteError: If we have an fquery and armageddon.
ReadOnlyPermissionDeniedError: If we try to alter the database and the
config is in readonly.
AlterError: Indirect raise by `_check_legality` or `where`.
AlreadyExecutedError: Indirect raise by `_check_legality` or `where`.
OneError: Indirect raise by `_check_legality` or `where`.
OtherThanWhereError: Indirect raise by `_check_legality` or `where`.
TerminalError: Indirect raise by `_check_legality` or `where`.
"""
if fquery and armageddon:
raise DeleteError("predicate")
# Can be test only with a config file and read_only = true,
# but this feature is test in:
# tests/algo/test_config.py with the test: test_readonly_true
if is_read_only(): # pragma: no cover
raise ReadOnlyPermissionDeniedError()
self._check_legality(Ct.DELETE)
self.flags.alter = True
self.cmd.append(Command(Ct.DELETE, DotMap(armageddon=armageddon)))
if fquery:
self.where(fquery)
return self
def insert(
self,
column: List[str] | str,
data: PyLinqSqlInsertType,
) -> SQLEnumerable:
"""
Add an INSERT command to the command list.
If data is a list, that means you make a multi insert.
For more see: `docs/doc/examples/examples.md`
or Usage.insert in the documentation.
Args:
column: Column where we want to insert.
data: Data we want to insert.
Returns:
Self.
Raises:
ReadOnlyPermissionDeniedError: If we try to alter the database and the
config is in readonly.
AlterError: Indirect raise by `_check_legality`.
AlreadyExecutedError: Indirect raise by `_check_legality`.
MoreThanZeroError: Indirect raise by `_check_legality`.
OneError: Indirect raise by `_check_legality`.
TerminalError: Indirect raise by `_check_legality`.
"""
# Can be test only with a config file and read_only = true,
# but this feature is test in:
# tests/algo/test_config.py with the test: test_readonly_true
if is_read_only(): # pragma: no cover
raise ReadOnlyPermissionDeniedError()
self._check_legality(Ct.INSERT)
self.flags.alter, self.flags.one = True, True
self.cmd.append(Command(Ct.INSERT, DotMap(data=data, column=column)))
return self
def simple_insert(self, **kwargs: dict[str, Any]) -> SQLEnumerable:
"""
Add an Insert command to the command list. Only for relationnal table.
You give all column name with the name of kwargs and value in args.
Args:
**kwargs: All data and name of the column is the key in kwargs.
Returns:
Self.
Raises:
ReadOnlyPermissionDeniedError: Indirect raise by `self.insert`.
AlterError: Indirect raise by `self.insert`.
AlreadyExecutedError: Indirect raise by `self.insert`.
MoreThanZeroError: Indirect raise by `self.insert`.
OneError: Indirect raise by `self.insert`.
TerminalError: Indirect raise by `self.insert`.
"""
return self.insert(column=list(kwargs.keys()), data=tuple(kwargs.values()))
def update(self, fquery: Callable[[MagicDotPath], MagicDotPath]) -> SQLEnumerable:
"""
Add an UPDATE command to the command list.
Args:
fquery: Lambda function to get path of the modification.
Returns:
Self.
Raises:
ReadOnlyPermissionDeniedError: If we try to alter the database and the
config is in readonly.
AlterError: Indirect raise by `_check_legality`.
AlreadyExecutedError: Indirect raise by `_check_legality`.
OneError: Indirect raise by `_check_legality`.
OtherThanWhereError: Indirect raise by `_check_legality`.
ReadOnlyPermissionDeniedError: Indirect raise by `_check_legality`.
TerminalError: Indirect raise by `_check_legality`.
"""
# Can be test only with a config file and read_only = true,
# but this feature is test in:
# tests/algo/test_config.py with the test: test_readonly_true
if is_read_only(): # pragma: no cover
raise ReadOnlyPermissionDeniedError()
self._check_legality(Ct.UPDATE)
self.flags.alter = True
self.cmd.append(Command(Ct.UPDATE, DotMap(fquery=fquery)))
return self
# -------------------
# | Select method |
# -------------------
def select(self, fquery: LambdaMagicDotPath | None = None) -> SQLEnumerable:
"""
Add an SELECT command to the command list.
Args:
fquery: Lambda function to get path(s) of selection. Its optional.
By default it's None for a request "SELECT *"
Returns:
Self.
Raises:
AlterError: Indirect raise by `_check_legality`.
AlreadyExecutedError: Indirect raise by `_check_legality`.
OneError: Indirect raise by `_check_legality`.
SelectError: Indirect raise by `_check_legality`.
TerminalError: Indirect raise by `_check_legality`.
"""
self._check_legality(Ct.SELECT)
self.flags.select = True
self.cmd.append(Command(Ct.SELECT, DotMap(fquery=fquery)))
return self
# -----------------
# | One methods |
# -----------------
def contains(
self,
fquery: Dict[str, Any] | Callable[[MagicDotPath], MagicDotPath],
) -> SQLEnumerable:
"""
Add an CONTAINS command to the command list.
Args:
fquery: Rawdata or path(s).
- Rawdata: Entire line of a table to know if the table contains this line.
- Path(s): Lambda function to get path(s) to know if a line of the table
contains this.
Returns:
Self.
Raises:
EmptyInputError: If fquery is a dict and is empty.
AlterError: Indirect raise by `_check_legality`.
AlreadyExecutedError: Indirect raise by `_check_legality`.
MoreThanZeroError: Indirect raise by `_check_legality`.
OneError: Indirect raise by `_check_legality`.
TerminalError: Indirect raise by `_check_legality`.
"""
self._check_legality(Ct.CONTAINS)
if isinstance(fquery, dict) and not fquery:
raise EmptyInputError("contains")
self.flags.one = True
self.cmd.append(Command(Ct.CONTAINS, DotMap(fquery=fquery)))
return self
def all(self, fquery: Callable[[MagicDotPath], MagicDotPath]) -> SQLEnumerable:
"""
Add an ALL command to the command list.
Args:
fquery: Lambda function to give the predicate.
Returns:
Self.
Raises:
AlterError: Indirect raise by `_check_legality`.
AlreadyExecutedError: Indirect raise by `_check_legality`.
MoreThanZeroError: Indirect raise by `_check_legality`.
OneError: Indirect raise by `_check_legality`.
TerminalError: Indirect raise by `_check_legality`.
"""
self._check_legality(Ct.ALL)
self.flags.one = True
self.cmd.append(Command(Ct.ALL, DotMap(fquery=fquery)))
return self
def any(
self,
fquery: Callable[[MagicDotPath], MagicDotPath] | None = None,
) -> SQLEnumerable:
"""
Add an ANY command to the command list.
Args:
fquery: Lambda function to give the predicate. Its optional.
Returns:
Self.
Raises:
AlterError: Indirect raise by `_check_legality`.
AlreadyExecutedError: Indirect raise by `_check_legality`.
MoreThanZeroError: Indirect raise by `_check_legality`.
OneError: Indirect raise by `_check_legality`.
TerminalError: Indirect raise by `_check_legality`.
"""
self._check_legality(Ct.ANY)
self.flags.one = True
self.cmd.append(Command(Ct.ANY, DotMap(fquery=fquery)))
return self
# ----------------------
# | Terminal methods |
# ----------------------
def count(self) -> SQLEnumerable:
"""
Add a COUNT command to the command list.
Returns:
Self.
Raises:
AlterError: Indirect raise by `_check_legality`.
AlreadyExecutedError: Indirect raise by `_check_legality`.
OneError: Indirect raise by `_check_legality`.
TerminalError: Indirect raise by `_check_legality`.
"""
self._check_legality(Ct.COUNT)
self.flags.terminal = Terminal.COUNT
self.cmd.append(Command(Ct.COUNT, None))
return self
def distinct(self) -> SQLEnumerable:
"""
Add a DISTINCT command to the command list.
Returns:
Self.
Raises:
AlterError: Indirect raise by `_check_legality`.
AlreadyExecutedError: Indirect raise by `_check_legality`.
OneError: Indirect raise by `_check_legality`.
TerminalError: Indirect raise by `_check_legality`.
"""
self._check_legality(Ct.DISTINCT)
self.flags.terminal = Terminal.DISTINCT
self.cmd.append(Command(Ct.DISTINCT, None))
return self
def element_at(self, index: int) -> SQLEnumerable:
"""
Add an ELEMENT_AT command to the command list.
Args:
index: Index of element we want. Must be a positive number.
Index must be strictly inferior than any other `take` call previously.
Returns:
Self.
Raises:
IndexError: If we make a `take(n)` before and index >= n.
AlterError: Indirect raise by `_check_legality`, `skip` or `take`.
AlreadyExecutedError: Indirect raise by `_check_legality`, `skip`
or `take`.
NegativeNumberError: Indirect raise by `skip` or `take`.
OneError: Indirect raise by `_check_legality`, `skip` or `take`.
SelectError: Indirect raise by `_check_legality`, `skip` or `take`.
TerminalError: Indirect raise by `_check_legality`, `skip` or `take`.
"""
commands = self.cmd
for cmd in commands:
if cmd.cmd_type == Ct.TAKE and index >= cmd.args.number:
raise IndexError()
self.skip(index).take(1)
self.flags.terminal = Terminal.ELEMENT_AT
return self
def element_at_or_default(self, index: int) -> SQLEnumerable:
"""
Add an ELEMENT_AT_OR_DEFAULT command to the command list.
Args:
index: Index of element we want. Must be a positive number.
Index must be strictly inferior than any other `take` call previously.
Returns:
Self.
Raises:
AlterError: Indirect raise by `element_at`.
AlreadyExecutedError: Indirect raise by `element_at`.
IndexError: Indirect raise by `element_at`.
NegativeNumberError: Indirect raise by `element_at`.
OneError: Indirect raise by `element_at`.
SelectError: Indirect raise by `element_at`.
TerminalError: Indirect raise by `element_at`.
"""
self.flags.default_cmd = True
return self.element_at(index)
def except_(self, exclude: SQLEnumerable) -> SQLEnumerable:
"""
Add an EXCEPT command to the command list.
Args:
exclude: SQLEnumerable to give the exclude predicate.
Returns:
Self.
Raises:
AlterError: Indirect raise by `_check_legality`.
AlreadyExecutedError: Indirect raise by `_check_legality`.
LengthMismatchError: Indirect raise by `get_command`.
NeedWhereError: Indirect raise by `get_command`.
OneError: Indirect raise by `_check_legality`.
psycopg.Error: Indirect raise by `get_command`.
TableError: Indirect raise by `get_command`.
TerminalError: Indirect raise by `_check_legality`.
TooManyReturnValueError: Indirect raise by `get_command`.
TypeError: Indirect raise by `get_command`.
TypeOperatorError: Indirect raise by `get_command`.
ValueError: Indirect raise by `get_command`.
"""
self._check_legality(Ct.EXCEPT_)
try:
self._check_select(exclude, "except")
except NeedSelectError:
pass
self.flags.terminal = Terminal.EXCEPT_
exclude_cmd = exclude.get_command()
self.cmd.append(Command(Ct.EXCEPT_, DotMap(exclude_cmd=exclude_cmd)))
return self
def first(self, fquery: LambdaMagicDotPath | None = None) -> SQLEnumerable:
"""
Add a FIRST command to the command list.
Args:
fquery: Lambda function to give the predicate. Its optional.
Returns:
Self.
Raises:
AlterError: Indirect raise by `element_at` or `where`.
AlreadyExecutedError: Indirect raise by `element_at` or `where`.
NegativeNumberError: Indirect raise by `element_at`.
OneError: Indirect raise by `element_at` or `where`.
SelectError: Indirect raise by `element_at` or `where`.
TerminalError: Indirect raise by `element_at` or `where`.
"""
if fquery:
self.where(fquery)
# element_at can be raise NegativeError but never with index == 0.
self.element_at(0)
return self
def first_or_default(
self,
fquery: LambdaMagicDotPath | None = None,
) -> SQLEnumerable:
"""
Add a FIRST_OR_DEFAULT command to the command list.
Args:
fquery: Lambda function to give the predicate. Its optional.
Returns:
Self.
Raises:
AlterError: Indirect raise by `first`.
AlreadyExecutedError: Indirect raise by `first`.
NegativeNumberError: Indirect raise by `first`.
OneError: Indirect raise by `first`.
SelectError: Indirect raise by `first`.
TerminalError: Indirect raise by `first`.
"""
self.flags.default_cmd = True
return self.first(fquery)
def group_by(
self,
by_fquery: LambdaMagicDotPath,
aggreg_fquery: LambdaMagicDotPathAggregate,
) -> SQLEnumerable:
"""
Add a GROUP_BY command to the command list.
Args:
by_fquery: lambda function to give the selection.
aggreg_fquery: lambda function to give the aggregation.
Returns:
Self.
Raises:
AlterError: Indirect raise by `_check_legality` or `select`.
AlreadyExecutedError: Indirect raise by `_check_legality` or `select`.
GroupByWithJoinError: Indirect raise by `_check_legality` or `select`.
OneError: Indirect raise by `_check_legality` or `select`.
SelectError: Indirect raise by `_check_legality` or `select`.
TerminalError: Indirect raise by `_check_legality` or `select`.
"""
self._check_legality(Ct.GROUP_BY)
self.select(by_fquery)
self.flags.terminal = Terminal.GROUP_BY
self.cmd.append(
Command(
Ct.GROUP_BY,
DotMap(
aggreg_fquery=aggreg_fquery,
),
),
)
return self
def intersect(self, sqle: SQLEnumerable) -> SQLEnumerable:
"""
Add an INTERSECT command to the command list.
Args:
sqle: An other SQLEnumerable to make the intersection.
Returns:
Self.
Raises:
AlterError: Indirect raise by `_check_legality`.
AlreadyExecutedError: Indirect raise by `_check_legality`.
DeleteError: Indirect raise by `build`.
LengthMismatchError: Indirect raise by `build`.
NeedSelectError: Indirect raise by `_check_legality`.
NeedWhereError: Indirect raise by `build`.
OneError: Indirect raise by `_check_legality`.
psycopg.Error: Indirect raise by `build`.
TableError: Indirect raise by `build`.
TerminalError: Indirect raise by `_check_legality`.
TooManyReturnValueError: Indirect raise by `build`.
TypeError: Indirect raise by `build`.
TypeOperatorError: Indirect raise by `build`.
ValueError: Indirect raise by `build`.
"""
self._check_legality(Ct.INTERSECT)
self._check_select(sqle, "intersect")
sqle_data = SQLEnumerableData(
sqle.connection,
sqle.flags,
sqle.cmd,
sqle.table,
sqle.length,
)
built_sqle_2 = build(sqle_data)
self.flags.terminal = Terminal.INTERSECT
self.cmd.append(Command(Ct.INTERSECT, DotMap(built_sqle_2=built_sqle_2)))
return self
def last(self, fquery: LambdaMagicDotPath | None = None) -> SQLEnumerable:
"""
Add a LAST command to the command list.
Args:
fquery: Lambda function to give the predicate. Its optional.
Returns:
Self.
Raises:
AlterError: Indirect raise by `_check_legality`, `where` or `take_last`.
AlreadyExecutedError: Indirect raise by `_check_legality`, `where`
or `take_last`.
or `take_last`.
OneError: Indirect raise by `_check_legality`, `where` or `take_last`.
or `take_last`.
SelectError: Indirect raise by `_check_legality`, `where` or `take_last`.
TerminalError: Indirect raise by `_check_legality`, `where` or `take_last`.
"""
self._check_legality(Ct.LAST)
if fquery:
self.where(fquery)
# `take_last` can raise NegativeNumberError
# but not here because wa always call take_last with 1.
self.take_last(1)
self.flags.terminal = Terminal.LAST
return self
def last_or_default(
self,
fquery: LambdaMagicDotPath | None = None,
) -> SQLEnumerable:
"""
Add a LAST_OR_DEFAULT command to the command list.
Args:
fquery: Lambda function to give the predicate. Its optional.
Returns:
Self.
Raises:
AlterError: Indirect raise by `last`.
AlreadyExecutedError: Indirect raise by `last`.
OneError: Indirect raise by `last`.
SelectError: Indirect raise by `last`.
TerminalError: Indirect raise by `last`.
"""
self.flags.default_cmd = True
return self.last(fquery)
def max(
self,
fquery: LambdaMagicDotPath | None,
cast_type: type | None = None,
) -> SQLEnumerable:
"""
Add a MAX command to the command list.
Args:
fquery: Lambda function to get the path(s) of the selection.
cast_type: Type in which we want to cast the path(s). Its optional.
Returns:
Self.
Raises:
AlterError: Indirect raise by `_check_legality`.
AlreadyExecutedError: Indirect raise by `_check_legality`.
NoMaxOrMinAfterLimitOffsetError: Indirect raise by `_check_legality`.
OneError: Indirect raise by `_check_legality`.
TerminalError: Indirect raise by `_check_legality`.
"""
self._check_legality(Ct.MAX)
self.flags.terminal = Terminal.MAX
self.cmd.append(Command(Ct.MAX, DotMap(fquery=fquery, cast_type=cast_type)))
return self
def min(
self,
fquery: LambdaMagicDotPath | None,
cast_type: type | None = None,
) -> SQLEnumerable:
"""
Add a MIN command to the command list.
Args:
fquery: Lambda function to get the path(s) of the selection.
cast_type: Type in which we want to cast the path(s). Its optional.
Returns:
Self.
Raises:
AlterError: Indirect raise by `_check_legality`.
AlreadyExecutedError: Indirect raise by `_check_legality`.
NoMaxOrMinAfterLimitOffsetError: Indirect raise by `_check_legality`.
OneError: Indirect raise by `_check_legality`.
TerminalError: Indirect raise by `_check_legality`.
"""
self._check_legality(Ct.MIN)
self.flags.terminal = Terminal.MIN
self.cmd.append(Command(Ct.MIN, DotMap(fquery=fquery, cast_type=cast_type)))
return self
def single(self, fquery: LambdaMagicDotPath | None = None) -> SQLEnumerable:
"""
Add a SINGLE command to the command list.
Args:
fquery: Lambda function to give the predicate. Its optional.
Returns:
Self.
Raises:
DeleteError: If we have an fquery and armageddon.
AlterError: Indirect raise by `_check_legality` or `where`.
AlreadyExecutedError: Indirect raise by `_check_legality` or `where`.
OneError: Indirect raise by `_check_legality` or `where`.
SelectError: Indirect raise by `_check_legality` or `where`.
TerminalError: Indirect raise by `_check_legality` or `where`.
"""
self._check_legality(Ct.SINGLE)
if fquery:
self.where(fquery)
self.flags.terminal = Terminal.SINGLE
return self
def single_or_default(
self,
fquery: LambdaMagicDotPath | None = None,
) -> SQLEnumerable:
"""
Add a SINGLE_OR_DEFAULT command to the command list.
Args:
fquery: Lambda function to give the predicate. Its optional.
Returns:
Self.
Raises:
AlterError: Indirect raise by `single`.
AlreadyExecutedError: Indirect raise by `single`.
OneError: Indirect raise by `single`.
SelectError: Indirect raise by `single`.
TerminalError: Indirect raise by `single`.
"""
self.flags.default_cmd = True
return self.single(fquery)
def union(self, sqle: SQLEnumerable, all_: bool = False) -> SQLEnumerable:
"""
Add an UNION command to the command list.
Args:
sqle: An other SQLEnumerable to make the union.
all_: Boolean to know if we want an UNION (False) or an UNION ALL (True).
By default: False.
Returns:
Self.
Raises:
AlterError: Indirect raise by `_check_legality`.
AlreadyExecutedError: Indirect raise by `_check_legality`.
DeleteError: Indirect raise by `build`.
LengthMismatchError: Indirect raise by `build`.
NeedSelectError: Indirect raise by `_check_legality`.
NeedWhereError: Indirect raise by `build`.
OneError: Indirect raise by `_check_legality`.
psycopg.Error: Indirect raise by `build`.
TableError: Indirect raise by `build`.
TerminalError: Indirect raise by `_check_legality`.
TooManyReturnValueError: Indirect raise by `build`.
TypeError: Indirect raise by `build`.
TypeOperatorError: Indirect raise by `build`.
ValueError: Indirect raise by `build`.
"""
self._check_legality(Ct.UNION)
self._check_select(sqle, "intersect")
sqle_data = SQLEnumerableData(
sqle.connection,
sqle.flags,
sqle.cmd,
sqle.table,
sqle.length,
)
built_sqle_2 = build(sqle_data)
self.flags.terminal = Terminal.UNION
self.cmd.append(Command(Ct.UNION, DotMap(built_sqle_2=built_sqle_2, all_=all_)))
return self
# ---------------------
# | Context methods |
# ---------------------
def group_join( # pylint: disable=too-many-arguments
self,
inner: SQLEnumerable,
outer_key: LambdaMagicDotPath,
inner_key: LambdaMagicDotPath,
result_function: LambdaMagicDotPathAggregate,
join_type: JoinType = JoinType.INNER,
) -> SQLEnumerable:
"""
Add a GROUP JOIN command to the command list.
Warning: For the moment you need to put inner before outer in the lambda
for example, if inner is satellite in the result function you need to make
lambda satellite, objects: stalellite.x.y, objects.a.b
Args:
inner: SQLEnumerable with which we want to make the join.
outer_key: Lambda function to get the path(s) parameters
for comparison in the outer SQLEnumerable.
inner_key: Lambda function to get the path(s) parameters
for comparison in the inner SQLEnumerable.
result_function: Lambda function to get the path(s) for the selection.
join_type: Type of join.
Returns:
Self.
Raises:
AlterError: Indirect raise by `_check_legality`.
AlreadyExecutedError: Indirect raise by `_check_legality`.
OneError: Indirect raise by `_check_legality`.
TerminalError: Indirect raise by `_check_legality`.
"""
self._check_legality(Ct.GROUP_JOIN)
outer = SQLEnumerableData(
self.connection,
self.flags,
self.cmd,
self.table,
self.length,
)
inner_sqled = SQLEnumerableData(
inner.connection,
inner.flags,
inner.cmd,
inner.table,
self.length,
)
self.flags.select, self.flags.join = True, True
self.cmd.append(
Command(
Ct.GROUP_JOIN,
DotMap(
outer=outer,
inner=inner_sqled,
outer_key=outer_key,
inner_key=inner_key,
result_function=result_function,
join_type=join_type,
),
),
)
return self
def join( # pylint: disable=too-many-arguments
self,
inner: SQLEnumerable,
outer_key: LambdaMagicDotPath,
inner_key: LambdaMagicDotPath,
result_function: LambdaMagicDotPath | None = None,
join_type: JoinType = JoinType.INNER,
) -> SQLEnumerable:
"""
Add a JOIN command to the command list.
Warning: For the moment you need to put inner before outer in the lambda
for example, if inner is satellite in the result function you need to make
lambda satellite, objects: stalellite.x.y, objects.a.b
Args:
inner: SQLEnumerable with which we want to make the join.
outer_key: Lambda function to get the path(s) parameters
for comparison in the outer SQLEnumerable.
inner_key: Lambda function to get the path(s) parameters
for comparison in the inner SQLEnumerable.
result_function: Lambda function to get the path(s) for the selection.
join_type: Type of join.
Returns:
Self.
Raises:
AlterError: Indirect raise by `_check_legality`.
AlreadyExecutedError: Indirect raise by `_check_legality`.
GroupByWithJoinError: Indirect raise by `_check_legality`.
OneError: Indirect raise by `_check_legality`.
TerminalError: Indirect raise by `_check_legality`.
"""
self._check_legality(Ct.JOIN)
outer = SQLEnumerableData(
self.connection,
self.flags,
self.cmd,
self.table,
self.length,
)
inner_sqled = SQLEnumerableData(
inner.connection,
inner.flags,
inner.cmd,
inner.table,
self.length,
)
self.flags.select, self.flags.join = True, True
self.cmd.append(
Command(
Ct.JOIN,
DotMap(
outer=outer,
inner=inner_sqled,
outer_key=outer_key,
inner_key=inner_key,
result_function=result_function,
join_type=join_type,
),
),
)
return self
def order_by(self, fquery: LambdaMagicDotPath) -> SQLEnumerable:
"""
Add an ORDER_BY command to the command list.
Args:
fquery: Lambda function to get the path(s) for the ordering.
Returns:
Self.
Raises:
AlterError: Indirect raise by `_check_legality`.
AlreadyExecutedError: Indirect raise by `_check_legality`.
OneError: Indirect raise by `_check_legality`.
TerminalError: Indirect raise by `_check_legality`.
"""
self._check_legality(Ct.ORDER_BY)
self.cmd.append(Command(Ct.ORDER_BY, DotMap(fquery=fquery)))
return self
def order_by_descending(self, fquery: LambdaMagicDotPath) -> SQLEnumerable:
"""
Add an ORDER_BY_DESCENDING command to the command list.
Args:
fquery: Lambda function to get the path(s) for the ordering.
Returns:
Self.
Raises:
AlterError: Indirect raise by `_check_legality`.
AlreadyExecutedError: Indirect raise by `_check_legality`.
OneError: Indirect raise by `_check_legality`.
TerminalError: Indirect raise by `_check_legality`.
"""
self._check_legality(Ct.ORDER_BY_DESC)
self.cmd.append(Command(Ct.ORDER_BY_DESC, DotMap(fquery=fquery)))
return self
def skip(self, number: int) -> SQLEnumerable:
"""
Add an SKIP command to the command list.
Args:
number: Number of element to skip.
Returns:
Self.
Raises:
NegativeNumberError: If number < 0.
AlterError: Indirect raise by `_check_legality`.
AlreadyExecutedError: Indirect raise by `_check_legality`.
OneError: Indirect raise by `_check_legality`.
TerminalError: Indirect raise by `_check_legality`.
"""
if number < 0:
raise NegativeNumberError()
self._check_legality(Ct.SKIP)
self.flags.limit_offset = True
self.cmd.append(Command(Ct.SKIP, DotMap(number=number)))
return self
def skip_last(self, number: int) -> SQLEnumerable:
"""
Add an SKIP_LAST command to the command list.
Args:
number: Number of element to skip the end.
Returns:
Self.
Raises:
NegativeNumberError: If number < 0.
AlterError: Indirect raise by `_check_legality`.
AlreadyExecutedError: Indirect raise by `_check_legality`.
OneError: Indirect raise by `_check_legality`.
TerminalError: Indirect raise by `_check_legality`.
"""
if number < 0:
raise NegativeNumberError()
self._check_legality(Ct.SKIP_LAST)
self.flags.limit_offset = True
self.cmd.append(Command(Ct.SKIP_LAST, DotMap(number=number)))
return self
def take(self, number: int) -> SQLEnumerable:
"""
Add an TAKE command to the command list.
Args:
number: Number of element to take.
Returns:
Self.
Raises:
NegativeNumberError: If number < 0.
AlterError: Indirect raise by `_check_legality`.
AlreadyExecutedError: Indirect raise by `_check_legality`.
OneError: Indirect raise by `_check_legality`.
TerminalError: Indirect raise by `_check_legality`.
"""
if number < 0:
raise NegativeNumberError()
self._check_legality(Ct.TAKE)
self.flags.limit_offset = True
self.cmd.append(Command(Ct.TAKE, DotMap(number=number)))
return self
def take_last(self, number: int) -> SQLEnumerable:
"""
Add an TAKE_LAST command to the command list.
Args:
number: Number of element to take.
Returns:
Self.
Raises:
NegativeNumberError: If number < 0.
AlterError: Indirect raise by `_check_legality`.
AlreadyExecutedError: Indirect raise by `_check_legality`.
OneError: Indirect raise by `_check_legality`.
TerminalError: Indirect raise by `_check_legality`.
"""
if number < 0:
raise NegativeNumberError()
self._check_legality(Ct.TAKE_LAST)
self.flags.limit_offset = True
self.cmd.append(Command(Ct.TAKE_LAST, DotMap(number=number)))
return self
def where(self, fquery: LambdaMagicDotPath) -> SQLEnumerable:
"""
Add an WHERE command to the command list.
Args:
fquery: Lambda function the get the path of the predicate.
Returns:
Self.
Raises:
AlterError: Indirect raise by `_check_legality`.
AlreadyExecutedError: Indirect raise by `_check_legality`.
OneError: Indirect raise by `_check_legality`.
TerminalError: Indirect raise by `_check_legality`.
"""
self._check_legality(Ct.WHERE)
self.cmd.append(Command(Ct.WHERE, DotMap(fquery=fquery)))
return self
def select_many(self) -> None:
"""
Generate a SELECT MANY request.
Raises:
NotImplementedError: Because it's functions is not implemented.
"""
raise NotImplementedError()
def to_list(self) -> None:
"""
Generate a TO LIST request.
Raises:
NotImplementedError: Because it's functions is not implemented.
"""
raise NotImplementedError()
def median(self) -> None:
"""
Generate a MEDIAN request.
Raises:
NotImplementedError: Because it's functions is not implemented.
"""
raise NotImplementedError()
def aggregate(self) -> None:
"""
Generate a AGGREGATE request.
Raises:
NotImplementedError: Because it's functions is not implemented.
"""
raise NotImplementedError()
def append(self) -> None:
"""
Generate a APPEND request.
Raises:
NotImplementedError: Because it's functions is not implemented.
"""
raise NotImplementedError()
def prepend(self) -> None:
"""
Generate a PREPEND request.
Raises:
NotImplementedError: Because it's functions is not implemented.
"""
raise NotImplementedError()
def empty(self) -> None:
"""
Generate a EMPTY request.
Raises:
NotImplementedError: Because it's functions is not implemented.
"""
raise NotImplementedError()
def range(self) -> None:
"""
Generate a RANGE request.
Raises:
NotImplementedError: Because it's functions is not implemented.
"""
raise NotImplementedError()
def repeat(self) -> None:
"""
Generate a REPEAT request.
Raises:
NotImplementedError: Because it's functions is not implemented.
"""
raise NotImplementedError()
def reverse(self) -> None:
"""
Generate a REVERSE request.
Raises:
NotImplementedError: Because it's functions is not implemented.
"""
raise NotImplementedError()
def skip_while(self) -> None:
"""
Generate a SKIP WHILE request.
Raises:
NotImplementedError: Because it's functions is not implemented.
"""
raise NotImplementedError()
def take_while(self) -> None:
"""
Generate a TAKE WHILE request.
Raises:
NotImplementedError: Because it's functions is not implemented.
"""
raise NotImplementedError()
def zip(self) -> None:
"""
Generate a ZIP request.
Raises:
NotImplementedError: Because it's functions is not implemented.
"""
raise NotImplementedError()
def default_if_empty(self) -> None:
"""
Generate a DEFAULT IF EMPTY request.
Raises:
NotImplementedError: Because it's functions is not implemented.
"""
raise NotImplementedError()
def execute(self) -> Enumerable | int | tuple | bool:
"""
Execute the request.
Returns:
Result of the execution. Can be an Enumerable, int, dict or boolean.
Raises:
AlreadyExecutedError: If this SQLEnumerable is already executed.
ActionError: If we don't make a `select`, an alter or a one command.
IndexError: If we make an `element_at` but not by default
and the command to execute is None.
EmptyQueryError: If cmd to execute is empty or Indirect raise by
`execute_select`, `execute_alter` or `execute_one`.
CursorCloseError: Indirect raise by
`execute_select`, `execute_alter` or `execute_one`.
DatabError: Indirect raise by `execute_select`.
DeleteError: Indirect raise by `get_command`.
EmptyRecord: Indirect raise by `execute_select`.
FetchError: Indirect raise by `execute_select`.
LengthMismatchError: Indirect raise by `get_command`.
NeedWhereError: Indirect raise by `get_command`.
psycopg.Error: Indirect raise by `get_command`.
TableError: Indirect raise by `get_command`.
TooManyReturnValueError: Indirect raise by `get_command`.
TypeError: Indirect raise by `get_command`.
TypeOperatorError: Indirect raise by `get_command`.
ValueError: Indirect raise by `get_command`.
"""
if self.executed:
raise AlreadyExecutedError()
if not self.flags.select and not self.flags.alter and not self.flags.one:
raise ActionError()
cmd_to_execute = self.get_command()
if not cmd_to_execute:
match (
self.flags.default_cmd,
self.flags.terminal,
self.flags.limit_offset,
):
case (True, _, _):
return None
case (False, Terminal.ELEMENT_AT, _):
raise IndexError() from ReturnEmptyEnumerable
case (_, _, True):
return Enumerable.empty()
case _:
# pylint disable=duplicate-code
raise EmptyQueryError(
ProgrammingError(),
cmd_to_execute,
"We can't execute empty request",
False,
) from ProgrammingError
# pylint enable=duplicate-code
hidden_sqle = SQLEnumerableData(
self.connection,
self.flags,
self.cmd,
self.table,
self.length,
)
if self.flags.select:
record = execute_select(cmd_to_execute, hidden_sqle)
elif self.flags.alter:
record = execute_alter(cmd_to_execute, hidden_sqle)
elif self.flags.one:
record = execute_one(cmd_to_execute, hidden_sqle)
self.executed = True
return record
__eq__(self, other)
special
Try equality between two SQL_Enumerable.
SQL_Enumerable_1 == SQL_Enumerable_2.
Source code in py_linq_sql/sql_enumerable/sql_enumerable.py
def __eq__(self, other: Any) -> bool:
"""
Try equality between two SQL_Enumerable.
SQL_Enumerable_1 == SQL_Enumerable_2.
"""
return equality(self, other)
__ne__(self, other)
special
Try no-equality between two SQL_Enumerable.
SQL_Enumerable_1 != SQL_Enumerable_2.
Source code in py_linq_sql/sql_enumerable/sql_enumerable.py
def __ne__(self, other: Any) -> bool:
"""
Try no-equality between two SQL_Enumerable.
SQL_Enumerable_1 != SQL_Enumerable_2.
"""
return bool(not self.__eq__(other))
__post_init__(self)
special
Initialize the command list and safe the table if it's an str.
Exceptions: |
|
---|
Source code in py_linq_sql/sql_enumerable/sql_enumerable.py
def __post_init__(self) -> None:
"""
Initialize the command list and safe the table if it's an str.
Raises:
psycopg.Error: Indirect raise by `safe`.
TablePermissionDeniedError: Indirect raise by
`is_valid_table_name_with_white_and_black_list`.
"""
if isinstance(self.table, str):
tables = self.table.split(".") if "." in self.table else [self.table]
safed_tables = []
for table in tables:
if not is_valid_table_name_with_white_and_black_list(table):
raise TablePermissionDeniedError(table)
safed_tables.append(safe(self.connection, table))
self.table = ".".join([f'"{stab[1:-1]}"' for stab in safed_tables])
self.cmd = []
aggregate(self)
Generate a AGGREGATE request.
Exceptions: |
|
---|
Source code in py_linq_sql/sql_enumerable/sql_enumerable.py
def aggregate(self) -> None:
"""
Generate a AGGREGATE request.
Raises:
NotImplementedError: Because it's functions is not implemented.
"""
raise NotImplementedError()
all(self, fquery)
Add an ALL command to the command list.
Parameters: |
|
---|
Returns: |
|
---|
Exceptions: |
|
---|
Source code in py_linq_sql/sql_enumerable/sql_enumerable.py
def all(self, fquery: Callable[[MagicDotPath], MagicDotPath]) -> SQLEnumerable:
"""
Add an ALL command to the command list.
Args:
fquery: Lambda function to give the predicate.
Returns:
Self.
Raises:
AlterError: Indirect raise by `_check_legality`.
AlreadyExecutedError: Indirect raise by `_check_legality`.
MoreThanZeroError: Indirect raise by `_check_legality`.
OneError: Indirect raise by `_check_legality`.
TerminalError: Indirect raise by `_check_legality`.
"""
self._check_legality(Ct.ALL)
self.flags.one = True
self.cmd.append(Command(Ct.ALL, DotMap(fquery=fquery)))
return self
any(self, fquery=None)
Add an ANY command to the command list.
Parameters: |
|
---|
Returns: |
|
---|
Exceptions: |
|
---|
Source code in py_linq_sql/sql_enumerable/sql_enumerable.py
def any(
self,
fquery: Callable[[MagicDotPath], MagicDotPath] | None = None,
) -> SQLEnumerable:
"""
Add an ANY command to the command list.
Args:
fquery: Lambda function to give the predicate. Its optional.
Returns:
Self.
Raises:
AlterError: Indirect raise by `_check_legality`.
AlreadyExecutedError: Indirect raise by `_check_legality`.
MoreThanZeroError: Indirect raise by `_check_legality`.
OneError: Indirect raise by `_check_legality`.
TerminalError: Indirect raise by `_check_legality`.
"""
self._check_legality(Ct.ANY)
self.flags.one = True
self.cmd.append(Command(Ct.ANY, DotMap(fquery=fquery)))
return self
append(self)
Generate a APPEND request.
Exceptions: |
|
---|
Source code in py_linq_sql/sql_enumerable/sql_enumerable.py
def append(self) -> None:
"""
Generate a APPEND request.
Raises:
NotImplementedError: Because it's functions is not implemented.
"""
raise NotImplementedError()
contains(self, fquery)
Add an CONTAINS command to the command list.
Parameters: |
|
---|
- Rawdata: Entire line of a table to know if the table contains this line.
- Path(s): Lambda function to get path(s) to know if a line of the table contains this.
Returns: |
|
---|
Exceptions: |
|
---|
Source code in py_linq_sql/sql_enumerable/sql_enumerable.py
def contains(
self,
fquery: Dict[str, Any] | Callable[[MagicDotPath], MagicDotPath],
) -> SQLEnumerable:
"""
Add an CONTAINS command to the command list.
Args:
fquery: Rawdata or path(s).
- Rawdata: Entire line of a table to know if the table contains this line.
- Path(s): Lambda function to get path(s) to know if a line of the table
contains this.
Returns:
Self.
Raises:
EmptyInputError: If fquery is a dict and is empty.
AlterError: Indirect raise by `_check_legality`.
AlreadyExecutedError: Indirect raise by `_check_legality`.
MoreThanZeroError: Indirect raise by `_check_legality`.
OneError: Indirect raise by `_check_legality`.
TerminalError: Indirect raise by `_check_legality`.
"""
self._check_legality(Ct.CONTAINS)
if isinstance(fquery, dict) and not fquery:
raise EmptyInputError("contains")
self.flags.one = True
self.cmd.append(Command(Ct.CONTAINS, DotMap(fquery=fquery)))
return self
copy(self)
Create a shallow copy of self.
Source code in py_linq_sql/sql_enumerable/sql_enumerable.py
def copy(self) -> SQLEnumerable:
"""Create a shallow copy of self."""
return SQLEnumerable(
self.connection,
self._format_table(),
self.flags.copy(),
self.executed,
self.length,
)
count(self)
Add a COUNT command to the command list.
Returns: |
|
---|
Exceptions: |
|
---|
Source code in py_linq_sql/sql_enumerable/sql_enumerable.py
def count(self) -> SQLEnumerable:
"""
Add a COUNT command to the command list.
Returns:
Self.
Raises:
AlterError: Indirect raise by `_check_legality`.
AlreadyExecutedError: Indirect raise by `_check_legality`.
OneError: Indirect raise by `_check_legality`.
TerminalError: Indirect raise by `_check_legality`.
"""
self._check_legality(Ct.COUNT)
self.flags.terminal = Terminal.COUNT
self.cmd.append(Command(Ct.COUNT, None))
return self
default_if_empty(self)
Generate a DEFAULT IF EMPTY request.
Exceptions: |
|
---|
Source code in py_linq_sql/sql_enumerable/sql_enumerable.py
def default_if_empty(self) -> None:
"""
Generate a DEFAULT IF EMPTY request.
Raises:
NotImplementedError: Because it's functions is not implemented.
"""
raise NotImplementedError()
delete(self, fquery=None, *, armageddon=False)
Add a DELETE command to the command list.
Parameters: |
|
---|
Returns: |
|
---|
Exceptions: |
|
---|
Source code in py_linq_sql/sql_enumerable/sql_enumerable.py
def delete(
self,
fquery: LambdaMagicDotPath | None = None,
*,
armageddon: bool = False,
) -> SQLEnumerable:
"""
Add a DELETE command to the command list.
Args:
fquery: Lambda function to get the element to delete.
armageddon: True if you want to delete ALL your table, by default is False.
Returns:
Self.
Raises:
DeleteError: If we have an fquery and armageddon.
ReadOnlyPermissionDeniedError: If we try to alter the database and the
config is in readonly.
AlterError: Indirect raise by `_check_legality` or `where`.
AlreadyExecutedError: Indirect raise by `_check_legality` or `where`.
OneError: Indirect raise by `_check_legality` or `where`.
OtherThanWhereError: Indirect raise by `_check_legality` or `where`.
TerminalError: Indirect raise by `_check_legality` or `where`.
"""
if fquery and armageddon:
raise DeleteError("predicate")
# Can be test only with a config file and read_only = true,
# but this feature is test in:
# tests/algo/test_config.py with the test: test_readonly_true
if is_read_only(): # pragma: no cover
raise ReadOnlyPermissionDeniedError()
self._check_legality(Ct.DELETE)
self.flags.alter = True
self.cmd.append(Command(Ct.DELETE, DotMap(armageddon=armageddon)))
if fquery:
self.where(fquery)
return self
distinct(self)
Add a DISTINCT command to the command list.
Returns: |
|
---|
Exceptions: |
|
---|
Source code in py_linq_sql/sql_enumerable/sql_enumerable.py
def distinct(self) -> SQLEnumerable:
"""
Add a DISTINCT command to the command list.
Returns:
Self.
Raises:
AlterError: Indirect raise by `_check_legality`.
AlreadyExecutedError: Indirect raise by `_check_legality`.
OneError: Indirect raise by `_check_legality`.
TerminalError: Indirect raise by `_check_legality`.
"""
self._check_legality(Ct.DISTINCT)
self.flags.terminal = Terminal.DISTINCT
self.cmd.append(Command(Ct.DISTINCT, None))
return self
element_at(self, index)
Add an ELEMENT_AT command to the command list.
Parameters: |
|
---|
Returns: |
|
---|
Exceptions: |
|
---|
Source code in py_linq_sql/sql_enumerable/sql_enumerable.py
def element_at(self, index: int) -> SQLEnumerable:
"""
Add an ELEMENT_AT command to the command list.
Args:
index: Index of element we want. Must be a positive number.
Index must be strictly inferior than any other `take` call previously.
Returns:
Self.
Raises:
IndexError: If we make a `take(n)` before and index >= n.
AlterError: Indirect raise by `_check_legality`, `skip` or `take`.
AlreadyExecutedError: Indirect raise by `_check_legality`, `skip`
or `take`.
NegativeNumberError: Indirect raise by `skip` or `take`.
OneError: Indirect raise by `_check_legality`, `skip` or `take`.
SelectError: Indirect raise by `_check_legality`, `skip` or `take`.
TerminalError: Indirect raise by `_check_legality`, `skip` or `take`.
"""
commands = self.cmd
for cmd in commands:
if cmd.cmd_type == Ct.TAKE and index >= cmd.args.number:
raise IndexError()
self.skip(index).take(1)
self.flags.terminal = Terminal.ELEMENT_AT
return self
element_at_or_default(self, index)
Add an ELEMENT_AT_OR_DEFAULT command to the command list.
Parameters: |
|
---|
Returns: |
|
---|
Exceptions: |
|
---|
Source code in py_linq_sql/sql_enumerable/sql_enumerable.py
def element_at_or_default(self, index: int) -> SQLEnumerable:
"""
Add an ELEMENT_AT_OR_DEFAULT command to the command list.
Args:
index: Index of element we want. Must be a positive number.
Index must be strictly inferior than any other `take` call previously.
Returns:
Self.
Raises:
AlterError: Indirect raise by `element_at`.
AlreadyExecutedError: Indirect raise by `element_at`.
IndexError: Indirect raise by `element_at`.
NegativeNumberError: Indirect raise by `element_at`.
OneError: Indirect raise by `element_at`.
SelectError: Indirect raise by `element_at`.
TerminalError: Indirect raise by `element_at`.
"""
self.flags.default_cmd = True
return self.element_at(index)
empty(self)
Generate a EMPTY request.
Exceptions: |
|
---|
Source code in py_linq_sql/sql_enumerable/sql_enumerable.py
def empty(self) -> None:
"""
Generate a EMPTY request.
Raises:
NotImplementedError: Because it's functions is not implemented.
"""
raise NotImplementedError()
except_(self, exclude)
Add an EXCEPT command to the command list.
Parameters: |
|
---|
Returns: |
|
---|
Exceptions: |
|
---|
Source code in py_linq_sql/sql_enumerable/sql_enumerable.py
def except_(self, exclude: SQLEnumerable) -> SQLEnumerable:
"""
Add an EXCEPT command to the command list.
Args:
exclude: SQLEnumerable to give the exclude predicate.
Returns:
Self.
Raises:
AlterError: Indirect raise by `_check_legality`.
AlreadyExecutedError: Indirect raise by `_check_legality`.
LengthMismatchError: Indirect raise by `get_command`.
NeedWhereError: Indirect raise by `get_command`.
OneError: Indirect raise by `_check_legality`.
psycopg.Error: Indirect raise by `get_command`.
TableError: Indirect raise by `get_command`.
TerminalError: Indirect raise by `_check_legality`.
TooManyReturnValueError: Indirect raise by `get_command`.
TypeError: Indirect raise by `get_command`.
TypeOperatorError: Indirect raise by `get_command`.
ValueError: Indirect raise by `get_command`.
"""
self._check_legality(Ct.EXCEPT_)
try:
self._check_select(exclude, "except")
except NeedSelectError:
pass
self.flags.terminal = Terminal.EXCEPT_
exclude_cmd = exclude.get_command()
self.cmd.append(Command(Ct.EXCEPT_, DotMap(exclude_cmd=exclude_cmd)))
return self
execute(self)
Execute the request.
Returns: |
|
---|
Exceptions: |
|
---|
Source code in py_linq_sql/sql_enumerable/sql_enumerable.py
def execute(self) -> Enumerable | int | tuple | bool:
"""
Execute the request.
Returns:
Result of the execution. Can be an Enumerable, int, dict or boolean.
Raises:
AlreadyExecutedError: If this SQLEnumerable is already executed.
ActionError: If we don't make a `select`, an alter or a one command.
IndexError: If we make an `element_at` but not by default
and the command to execute is None.
EmptyQueryError: If cmd to execute is empty or Indirect raise by
`execute_select`, `execute_alter` or `execute_one`.
CursorCloseError: Indirect raise by
`execute_select`, `execute_alter` or `execute_one`.
DatabError: Indirect raise by `execute_select`.
DeleteError: Indirect raise by `get_command`.
EmptyRecord: Indirect raise by `execute_select`.
FetchError: Indirect raise by `execute_select`.
LengthMismatchError: Indirect raise by `get_command`.
NeedWhereError: Indirect raise by `get_command`.
psycopg.Error: Indirect raise by `get_command`.
TableError: Indirect raise by `get_command`.
TooManyReturnValueError: Indirect raise by `get_command`.
TypeError: Indirect raise by `get_command`.
TypeOperatorError: Indirect raise by `get_command`.
ValueError: Indirect raise by `get_command`.
"""
if self.executed:
raise AlreadyExecutedError()
if not self.flags.select and not self.flags.alter and not self.flags.one:
raise ActionError()
cmd_to_execute = self.get_command()
if not cmd_to_execute:
match (
self.flags.default_cmd,
self.flags.terminal,
self.flags.limit_offset,
):
case (True, _, _):
return None
case (False, Terminal.ELEMENT_AT, _):
raise IndexError() from ReturnEmptyEnumerable
case (_, _, True):
return Enumerable.empty()
case _:
# pylint disable=duplicate-code
raise EmptyQueryError(
ProgrammingError(),
cmd_to_execute,
"We can't execute empty request",
False,
) from ProgrammingError
# pylint enable=duplicate-code
hidden_sqle = SQLEnumerableData(
self.connection,
self.flags,
self.cmd,
self.table,
self.length,
)
if self.flags.select:
record = execute_select(cmd_to_execute, hidden_sqle)
elif self.flags.alter:
record = execute_alter(cmd_to_execute, hidden_sqle)
elif self.flags.one:
record = execute_one(cmd_to_execute, hidden_sqle)
self.executed = True
return record
first(self, fquery=None)
Add a FIRST command to the command list.
Parameters: |
|
---|
Returns: |
|
---|
Exceptions: |
|
---|
Source code in py_linq_sql/sql_enumerable/sql_enumerable.py
def first(self, fquery: LambdaMagicDotPath | None = None) -> SQLEnumerable:
"""
Add a FIRST command to the command list.
Args:
fquery: Lambda function to give the predicate. Its optional.
Returns:
Self.
Raises:
AlterError: Indirect raise by `element_at` or `where`.
AlreadyExecutedError: Indirect raise by `element_at` or `where`.
NegativeNumberError: Indirect raise by `element_at`.
OneError: Indirect raise by `element_at` or `where`.
SelectError: Indirect raise by `element_at` or `where`.
TerminalError: Indirect raise by `element_at` or `where`.
"""
if fquery:
self.where(fquery)
# element_at can be raise NegativeError but never with index == 0.
self.element_at(0)
return self
first_or_default(self, fquery=None)
Add a FIRST_OR_DEFAULT command to the command list.
Parameters: |
|
---|
Returns: |
|
---|
Exceptions: |
|
---|
Source code in py_linq_sql/sql_enumerable/sql_enumerable.py
def first_or_default(
self,
fquery: LambdaMagicDotPath | None = None,
) -> SQLEnumerable:
"""
Add a FIRST_OR_DEFAULT command to the command list.
Args:
fquery: Lambda function to give the predicate. Its optional.
Returns:
Self.
Raises:
AlterError: Indirect raise by `first`.
AlreadyExecutedError: Indirect raise by `first`.
NegativeNumberError: Indirect raise by `first`.
OneError: Indirect raise by `first`.
SelectError: Indirect raise by `first`.
TerminalError: Indirect raise by `first`.
"""
self.flags.default_cmd = True
return self.first(fquery)
get_command(self)
Get the command as an str.
Returns: |
|
---|
Exceptions: |
|
---|
Source code in py_linq_sql/sql_enumerable/sql_enumerable.py
def get_command(self) -> str | None:
"""
Get the command as an str.
Returns:
The request as an str.
Raises:
DeleteError: Indirect raise by `build` or `_set_count`.
LengthMismatchError: Indirect raise by `build` or `_set_count`.
NeedWhereError: Indirect raise by `build` or `_set_count`.
psycopg.Error: Indirect raise by `build` or `_set_count`.
TableError: Indirect raise by `build` or `_set_count`.
TooManyReturnValueError: Indirect raise by `build` or `_set_count`.
TypeError: Indirect raise by `build` or `_set_count`.
TypeOperatorError: Indirect raise by `build` or `_set_count`.
UnknownCommandTypeError: Indirect raise by `build` or `_set_count`.
ValueError: Indirect raise by `build` or `_set_count`.
"""
if self.flags.limit_offset:
self._set_count()
hidden_sqle = SQLEnumerableData(
self.connection,
self.flags,
self.cmd,
self.table,
self.length,
)
try:
return build(hidden_sqle)
except ReturnEmptyEnumerable:
return None
group_by(self, by_fquery, aggreg_fquery)
Add a GROUP_BY command to the command list.
Parameters: |
|
---|
Returns: |
|
---|
Exceptions: |
|
---|
Source code in py_linq_sql/sql_enumerable/sql_enumerable.py
def group_by(
self,
by_fquery: LambdaMagicDotPath,
aggreg_fquery: LambdaMagicDotPathAggregate,
) -> SQLEnumerable:
"""
Add a GROUP_BY command to the command list.
Args:
by_fquery: lambda function to give the selection.
aggreg_fquery: lambda function to give the aggregation.
Returns:
Self.
Raises:
AlterError: Indirect raise by `_check_legality` or `select`.
AlreadyExecutedError: Indirect raise by `_check_legality` or `select`.
GroupByWithJoinError: Indirect raise by `_check_legality` or `select`.
OneError: Indirect raise by `_check_legality` or `select`.
SelectError: Indirect raise by `_check_legality` or `select`.
TerminalError: Indirect raise by `_check_legality` or `select`.
"""
self._check_legality(Ct.GROUP_BY)
self.select(by_fquery)
self.flags.terminal = Terminal.GROUP_BY
self.cmd.append(
Command(
Ct.GROUP_BY,
DotMap(
aggreg_fquery=aggreg_fquery,
),
),
)
return self
group_join(self, inner, outer_key, inner_key, result_function, join_type=<JoinType.INNER: _Type(is_intersect=False, as_str='INNER')>)
Add a GROUP JOIN command to the command list.
Warning: For the moment you need to put inner before outer in the lambda
for example, if inner is satellite in the result function you need to make lambda satellite, objects: stalellite.x.y, objects.a.b
Parameters: |
|
---|
Returns: |
|
---|
Exceptions: |
|
---|
Source code in py_linq_sql/sql_enumerable/sql_enumerable.py
def group_join( # pylint: disable=too-many-arguments
self,
inner: SQLEnumerable,
outer_key: LambdaMagicDotPath,
inner_key: LambdaMagicDotPath,
result_function: LambdaMagicDotPathAggregate,
join_type: JoinType = JoinType.INNER,
) -> SQLEnumerable:
"""
Add a GROUP JOIN command to the command list.
Warning: For the moment you need to put inner before outer in the lambda
for example, if inner is satellite in the result function you need to make
lambda satellite, objects: stalellite.x.y, objects.a.b
Args:
inner: SQLEnumerable with which we want to make the join.
outer_key: Lambda function to get the path(s) parameters
for comparison in the outer SQLEnumerable.
inner_key: Lambda function to get the path(s) parameters
for comparison in the inner SQLEnumerable.
result_function: Lambda function to get the path(s) for the selection.
join_type: Type of join.
Returns:
Self.
Raises:
AlterError: Indirect raise by `_check_legality`.
AlreadyExecutedError: Indirect raise by `_check_legality`.
OneError: Indirect raise by `_check_legality`.
TerminalError: Indirect raise by `_check_legality`.
"""
self._check_legality(Ct.GROUP_JOIN)
outer = SQLEnumerableData(
self.connection,
self.flags,
self.cmd,
self.table,
self.length,
)
inner_sqled = SQLEnumerableData(
inner.connection,
inner.flags,
inner.cmd,
inner.table,
self.length,
)
self.flags.select, self.flags.join = True, True
self.cmd.append(
Command(
Ct.GROUP_JOIN,
DotMap(
outer=outer,
inner=inner_sqled,
outer_key=outer_key,
inner_key=inner_key,
result_function=result_function,
join_type=join_type,
),
),
)
return self
insert(self, column, data)
Add an INSERT command to the command list.
If data is a list, that means you make a multi insert.
For more see: docs/doc/examples/examples.md
or Usage.insert in the documentation.
Parameters: |
|
---|
Returns: |
|
---|
Exceptions: |
|
---|
Source code in py_linq_sql/sql_enumerable/sql_enumerable.py
def insert(
self,
column: List[str] | str,
data: PyLinqSqlInsertType,
) -> SQLEnumerable:
"""
Add an INSERT command to the command list.
If data is a list, that means you make a multi insert.
For more see: `docs/doc/examples/examples.md`
or Usage.insert in the documentation.
Args:
column: Column where we want to insert.
data: Data we want to insert.
Returns:
Self.
Raises:
ReadOnlyPermissionDeniedError: If we try to alter the database and the
config is in readonly.
AlterError: Indirect raise by `_check_legality`.
AlreadyExecutedError: Indirect raise by `_check_legality`.
MoreThanZeroError: Indirect raise by `_check_legality`.
OneError: Indirect raise by `_check_legality`.
TerminalError: Indirect raise by `_check_legality`.
"""
# Can be test only with a config file and read_only = true,
# but this feature is test in:
# tests/algo/test_config.py with the test: test_readonly_true
if is_read_only(): # pragma: no cover
raise ReadOnlyPermissionDeniedError()
self._check_legality(Ct.INSERT)
self.flags.alter, self.flags.one = True, True
self.cmd.append(Command(Ct.INSERT, DotMap(data=data, column=column)))
return self
intersect(self, sqle)
Add an INTERSECT command to the command list.
Parameters: |
|
---|
Returns: |
|
---|
Exceptions: |
|
---|
Source code in py_linq_sql/sql_enumerable/sql_enumerable.py
def intersect(self, sqle: SQLEnumerable) -> SQLEnumerable:
"""
Add an INTERSECT command to the command list.
Args:
sqle: An other SQLEnumerable to make the intersection.
Returns:
Self.
Raises:
AlterError: Indirect raise by `_check_legality`.
AlreadyExecutedError: Indirect raise by `_check_legality`.
DeleteError: Indirect raise by `build`.
LengthMismatchError: Indirect raise by `build`.
NeedSelectError: Indirect raise by `_check_legality`.
NeedWhereError: Indirect raise by `build`.
OneError: Indirect raise by `_check_legality`.
psycopg.Error: Indirect raise by `build`.
TableError: Indirect raise by `build`.
TerminalError: Indirect raise by `_check_legality`.
TooManyReturnValueError: Indirect raise by `build`.
TypeError: Indirect raise by `build`.
TypeOperatorError: Indirect raise by `build`.
ValueError: Indirect raise by `build`.
"""
self._check_legality(Ct.INTERSECT)
self._check_select(sqle, "intersect")
sqle_data = SQLEnumerableData(
sqle.connection,
sqle.flags,
sqle.cmd,
sqle.table,
sqle.length,
)
built_sqle_2 = build(sqle_data)
self.flags.terminal = Terminal.INTERSECT
self.cmd.append(Command(Ct.INTERSECT, DotMap(built_sqle_2=built_sqle_2)))
return self
join(self, inner, outer_key, inner_key, result_function=None, join_type=<JoinType.INNER: _Type(is_intersect=False, as_str='INNER')>)
Add a JOIN command to the command list.
Warning: For the moment you need to put inner before outer in the lambda
for example, if inner is satellite in the result function you need to make lambda satellite, objects: stalellite.x.y, objects.a.b
Parameters: |
|
---|
Returns: |
|
---|
Exceptions: |
|
---|
Source code in py_linq_sql/sql_enumerable/sql_enumerable.py
def join( # pylint: disable=too-many-arguments
self,
inner: SQLEnumerable,
outer_key: LambdaMagicDotPath,
inner_key: LambdaMagicDotPath,
result_function: LambdaMagicDotPath | None = None,
join_type: JoinType = JoinType.INNER,
) -> SQLEnumerable:
"""
Add a JOIN command to the command list.
Warning: For the moment you need to put inner before outer in the lambda
for example, if inner is satellite in the result function you need to make
lambda satellite, objects: stalellite.x.y, objects.a.b
Args:
inner: SQLEnumerable with which we want to make the join.
outer_key: Lambda function to get the path(s) parameters
for comparison in the outer SQLEnumerable.
inner_key: Lambda function to get the path(s) parameters
for comparison in the inner SQLEnumerable.
result_function: Lambda function to get the path(s) for the selection.
join_type: Type of join.
Returns:
Self.
Raises:
AlterError: Indirect raise by `_check_legality`.
AlreadyExecutedError: Indirect raise by `_check_legality`.
GroupByWithJoinError: Indirect raise by `_check_legality`.
OneError: Indirect raise by `_check_legality`.
TerminalError: Indirect raise by `_check_legality`.
"""
self._check_legality(Ct.JOIN)
outer = SQLEnumerableData(
self.connection,
self.flags,
self.cmd,
self.table,
self.length,
)
inner_sqled = SQLEnumerableData(
inner.connection,
inner.flags,
inner.cmd,
inner.table,
self.length,
)
self.flags.select, self.flags.join = True, True
self.cmd.append(
Command(
Ct.JOIN,
DotMap(
outer=outer,
inner=inner_sqled,
outer_key=outer_key,
inner_key=inner_key,
result_function=result_function,
join_type=join_type,
),
),
)
return self
last(self, fquery=None)
Add a LAST command to the command list.
Parameters: |
|
---|
Returns: |
|
---|
Exceptions: |
|
---|
Source code in py_linq_sql/sql_enumerable/sql_enumerable.py
def last(self, fquery: LambdaMagicDotPath | None = None) -> SQLEnumerable:
"""
Add a LAST command to the command list.
Args:
fquery: Lambda function to give the predicate. Its optional.
Returns:
Self.
Raises:
AlterError: Indirect raise by `_check_legality`, `where` or `take_last`.
AlreadyExecutedError: Indirect raise by `_check_legality`, `where`
or `take_last`.
or `take_last`.
OneError: Indirect raise by `_check_legality`, `where` or `take_last`.
or `take_last`.
SelectError: Indirect raise by `_check_legality`, `where` or `take_last`.
TerminalError: Indirect raise by `_check_legality`, `where` or `take_last`.
"""
self._check_legality(Ct.LAST)
if fquery:
self.where(fquery)
# `take_last` can raise NegativeNumberError
# but not here because wa always call take_last with 1.
self.take_last(1)
self.flags.terminal = Terminal.LAST
return self
last_or_default(self, fquery=None)
Add a LAST_OR_DEFAULT command to the command list.
Parameters: |
|
---|
Returns: |
|
---|
Exceptions: |
|
---|
Source code in py_linq_sql/sql_enumerable/sql_enumerable.py
def last_or_default(
self,
fquery: LambdaMagicDotPath | None = None,
) -> SQLEnumerable:
"""
Add a LAST_OR_DEFAULT command to the command list.
Args:
fquery: Lambda function to give the predicate. Its optional.
Returns:
Self.
Raises:
AlterError: Indirect raise by `last`.
AlreadyExecutedError: Indirect raise by `last`.
OneError: Indirect raise by `last`.
SelectError: Indirect raise by `last`.
TerminalError: Indirect raise by `last`.
"""
self.flags.default_cmd = True
return self.last(fquery)
max(self, fquery, cast_type=None)
Add a MAX command to the command list.
Parameters: |
|
---|
Returns: |
|
---|
Exceptions: |
|
---|
Source code in py_linq_sql/sql_enumerable/sql_enumerable.py
def max(
self,
fquery: LambdaMagicDotPath | None,
cast_type: type | None = None,
) -> SQLEnumerable:
"""
Add a MAX command to the command list.
Args:
fquery: Lambda function to get the path(s) of the selection.
cast_type: Type in which we want to cast the path(s). Its optional.
Returns:
Self.
Raises:
AlterError: Indirect raise by `_check_legality`.
AlreadyExecutedError: Indirect raise by `_check_legality`.
NoMaxOrMinAfterLimitOffsetError: Indirect raise by `_check_legality`.
OneError: Indirect raise by `_check_legality`.
TerminalError: Indirect raise by `_check_legality`.
"""
self._check_legality(Ct.MAX)
self.flags.terminal = Terminal.MAX
self.cmd.append(Command(Ct.MAX, DotMap(fquery=fquery, cast_type=cast_type)))
return self
median(self)
Generate a MEDIAN request.
Exceptions: |
|
---|
Source code in py_linq_sql/sql_enumerable/sql_enumerable.py
def median(self) -> None:
"""
Generate a MEDIAN request.
Raises:
NotImplementedError: Because it's functions is not implemented.
"""
raise NotImplementedError()
min(self, fquery, cast_type=None)
Add a MIN command to the command list.
Parameters: |
|
---|
Returns: |
|
---|
Exceptions: |
|
---|
Source code in py_linq_sql/sql_enumerable/sql_enumerable.py
def min(
self,
fquery: LambdaMagicDotPath | None,
cast_type: type | None = None,
) -> SQLEnumerable:
"""
Add a MIN command to the command list.
Args:
fquery: Lambda function to get the path(s) of the selection.
cast_type: Type in which we want to cast the path(s). Its optional.
Returns:
Self.
Raises:
AlterError: Indirect raise by `_check_legality`.
AlreadyExecutedError: Indirect raise by `_check_legality`.
NoMaxOrMinAfterLimitOffsetError: Indirect raise by `_check_legality`.
OneError: Indirect raise by `_check_legality`.
TerminalError: Indirect raise by `_check_legality`.
"""
self._check_legality(Ct.MIN)
self.flags.terminal = Terminal.MIN
self.cmd.append(Command(Ct.MIN, DotMap(fquery=fquery, cast_type=cast_type)))
return self
order_by(self, fquery)
Add an ORDER_BY command to the command list.
Parameters: |
|
---|
Returns: |
|
---|
Exceptions: |
|
---|
Source code in py_linq_sql/sql_enumerable/sql_enumerable.py
def order_by(self, fquery: LambdaMagicDotPath) -> SQLEnumerable:
"""
Add an ORDER_BY command to the command list.
Args:
fquery: Lambda function to get the path(s) for the ordering.
Returns:
Self.
Raises:
AlterError: Indirect raise by `_check_legality`.
AlreadyExecutedError: Indirect raise by `_check_legality`.
OneError: Indirect raise by `_check_legality`.
TerminalError: Indirect raise by `_check_legality`.
"""
self._check_legality(Ct.ORDER_BY)
self.cmd.append(Command(Ct.ORDER_BY, DotMap(fquery=fquery)))
return self
order_by_descending(self, fquery)
Add an ORDER_BY_DESCENDING command to the command list.
Parameters: |
|
---|
Returns: |
|
---|
Exceptions: |
|
---|
Source code in py_linq_sql/sql_enumerable/sql_enumerable.py
def order_by_descending(self, fquery: LambdaMagicDotPath) -> SQLEnumerable:
"""
Add an ORDER_BY_DESCENDING command to the command list.
Args:
fquery: Lambda function to get the path(s) for the ordering.
Returns:
Self.
Raises:
AlterError: Indirect raise by `_check_legality`.
AlreadyExecutedError: Indirect raise by `_check_legality`.
OneError: Indirect raise by `_check_legality`.
TerminalError: Indirect raise by `_check_legality`.
"""
self._check_legality(Ct.ORDER_BY_DESC)
self.cmd.append(Command(Ct.ORDER_BY_DESC, DotMap(fquery=fquery)))
return self
prepend(self)
Generate a PREPEND request.
Exceptions: |
|
---|
Source code in py_linq_sql/sql_enumerable/sql_enumerable.py
def prepend(self) -> None:
"""
Generate a PREPEND request.
Raises:
NotImplementedError: Because it's functions is not implemented.
"""
raise NotImplementedError()
range(self)
Generate a RANGE request.
Exceptions: |
|
---|
Source code in py_linq_sql/sql_enumerable/sql_enumerable.py
def range(self) -> None:
"""
Generate a RANGE request.
Raises:
NotImplementedError: Because it's functions is not implemented.
"""
raise NotImplementedError()
repeat(self)
Generate a REPEAT request.
Exceptions: |
|
---|
Source code in py_linq_sql/sql_enumerable/sql_enumerable.py
def repeat(self) -> None:
"""
Generate a REPEAT request.
Raises:
NotImplementedError: Because it's functions is not implemented.
"""
raise NotImplementedError()
reverse(self)
Generate a REVERSE request.
Exceptions: |
|
---|
Source code in py_linq_sql/sql_enumerable/sql_enumerable.py
def reverse(self) -> None:
"""
Generate a REVERSE request.
Raises:
NotImplementedError: Because it's functions is not implemented.
"""
raise NotImplementedError()
select(self, fquery=None)
Add an SELECT command to the command list.
Parameters: |
|
---|
Returns: |
|
---|
Exceptions: |
|
---|
Source code in py_linq_sql/sql_enumerable/sql_enumerable.py
def select(self, fquery: LambdaMagicDotPath | None = None) -> SQLEnumerable:
"""
Add an SELECT command to the command list.
Args:
fquery: Lambda function to get path(s) of selection. Its optional.
By default it's None for a request "SELECT *"
Returns:
Self.
Raises:
AlterError: Indirect raise by `_check_legality`.
AlreadyExecutedError: Indirect raise by `_check_legality`.
OneError: Indirect raise by `_check_legality`.
SelectError: Indirect raise by `_check_legality`.
TerminalError: Indirect raise by `_check_legality`.
"""
self._check_legality(Ct.SELECT)
self.flags.select = True
self.cmd.append(Command(Ct.SELECT, DotMap(fquery=fquery)))
return self
select_many(self)
Generate a SELECT MANY request.
Exceptions: |
|
---|
Source code in py_linq_sql/sql_enumerable/sql_enumerable.py
def select_many(self) -> None:
"""
Generate a SELECT MANY request.
Raises:
NotImplementedError: Because it's functions is not implemented.
"""
raise NotImplementedError()
simple_insert(self, **kwargs)
Add an Insert command to the command list. Only for relationnal table.
You give all column name with the name of kwargs and value in args.
Parameters: |
|
---|
Returns: |
|
---|
Exceptions: |
|
---|
Source code in py_linq_sql/sql_enumerable/sql_enumerable.py
def simple_insert(self, **kwargs: dict[str, Any]) -> SQLEnumerable:
"""
Add an Insert command to the command list. Only for relationnal table.
You give all column name with the name of kwargs and value in args.
Args:
**kwargs: All data and name of the column is the key in kwargs.
Returns:
Self.
Raises:
ReadOnlyPermissionDeniedError: Indirect raise by `self.insert`.
AlterError: Indirect raise by `self.insert`.
AlreadyExecutedError: Indirect raise by `self.insert`.
MoreThanZeroError: Indirect raise by `self.insert`.
OneError: Indirect raise by `self.insert`.
TerminalError: Indirect raise by `self.insert`.
"""
return self.insert(column=list(kwargs.keys()), data=tuple(kwargs.values()))
single(self, fquery=None)
Add a SINGLE command to the command list.
Parameters: |
|
---|
Returns: |
|
---|
Exceptions: |
|
---|
Source code in py_linq_sql/sql_enumerable/sql_enumerable.py
def single(self, fquery: LambdaMagicDotPath | None = None) -> SQLEnumerable:
"""
Add a SINGLE command to the command list.
Args:
fquery: Lambda function to give the predicate. Its optional.
Returns:
Self.
Raises:
DeleteError: If we have an fquery and armageddon.
AlterError: Indirect raise by `_check_legality` or `where`.
AlreadyExecutedError: Indirect raise by `_check_legality` or `where`.
OneError: Indirect raise by `_check_legality` or `where`.
SelectError: Indirect raise by `_check_legality` or `where`.
TerminalError: Indirect raise by `_check_legality` or `where`.
"""
self._check_legality(Ct.SINGLE)
if fquery:
self.where(fquery)
self.flags.terminal = Terminal.SINGLE
return self
single_or_default(self, fquery=None)
Add a SINGLE_OR_DEFAULT command to the command list.
Parameters: |
|
---|
Returns: |
|
---|
Exceptions: |
|
---|
Source code in py_linq_sql/sql_enumerable/sql_enumerable.py
def single_or_default(
self,
fquery: LambdaMagicDotPath | None = None,
) -> SQLEnumerable:
"""
Add a SINGLE_OR_DEFAULT command to the command list.
Args:
fquery: Lambda function to give the predicate. Its optional.
Returns:
Self.
Raises:
AlterError: Indirect raise by `single`.
AlreadyExecutedError: Indirect raise by `single`.
OneError: Indirect raise by `single`.
SelectError: Indirect raise by `single`.
TerminalError: Indirect raise by `single`.
"""
self.flags.default_cmd = True
return self.single(fquery)
skip(self, number)
Add an SKIP command to the command list.
Parameters: |
|
---|
Returns: |
|
---|
Exceptions: |
|
---|
Source code in py_linq_sql/sql_enumerable/sql_enumerable.py
def skip(self, number: int) -> SQLEnumerable:
"""
Add an SKIP command to the command list.
Args:
number: Number of element to skip.
Returns:
Self.
Raises:
NegativeNumberError: If number < 0.
AlterError: Indirect raise by `_check_legality`.
AlreadyExecutedError: Indirect raise by `_check_legality`.
OneError: Indirect raise by `_check_legality`.
TerminalError: Indirect raise by `_check_legality`.
"""
if number < 0:
raise NegativeNumberError()
self._check_legality(Ct.SKIP)
self.flags.limit_offset = True
self.cmd.append(Command(Ct.SKIP, DotMap(number=number)))
return self
skip_last(self, number)
Add an SKIP_LAST command to the command list.
Parameters: |
|
---|
Returns: |
|
---|
Exceptions: |
|
---|
Source code in py_linq_sql/sql_enumerable/sql_enumerable.py
def skip_last(self, number: int) -> SQLEnumerable:
"""
Add an SKIP_LAST command to the command list.
Args:
number: Number of element to skip the end.
Returns:
Self.
Raises:
NegativeNumberError: If number < 0.
AlterError: Indirect raise by `_check_legality`.
AlreadyExecutedError: Indirect raise by `_check_legality`.
OneError: Indirect raise by `_check_legality`.
TerminalError: Indirect raise by `_check_legality`.
"""
if number < 0:
raise NegativeNumberError()
self._check_legality(Ct.SKIP_LAST)
self.flags.limit_offset = True
self.cmd.append(Command(Ct.SKIP_LAST, DotMap(number=number)))
return self
skip_while(self)
Generate a SKIP WHILE request.
Exceptions: |
|
---|
Source code in py_linq_sql/sql_enumerable/sql_enumerable.py
def skip_while(self) -> None:
"""
Generate a SKIP WHILE request.
Raises:
NotImplementedError: Because it's functions is not implemented.
"""
raise NotImplementedError()
take(self, number)
Add an TAKE command to the command list.
Parameters: |
|
---|
Returns: |
|
---|
Exceptions: |
|
---|
Source code in py_linq_sql/sql_enumerable/sql_enumerable.py
def take(self, number: int) -> SQLEnumerable:
"""
Add an TAKE command to the command list.
Args:
number: Number of element to take.
Returns:
Self.
Raises:
NegativeNumberError: If number < 0.
AlterError: Indirect raise by `_check_legality`.
AlreadyExecutedError: Indirect raise by `_check_legality`.
OneError: Indirect raise by `_check_legality`.
TerminalError: Indirect raise by `_check_legality`.
"""
if number < 0:
raise NegativeNumberError()
self._check_legality(Ct.TAKE)
self.flags.limit_offset = True
self.cmd.append(Command(Ct.TAKE, DotMap(number=number)))
return self
take_last(self, number)
Add an TAKE_LAST command to the command list.
Parameters: |
|
---|
Returns: |
|
---|
Exceptions: |
|
---|
Source code in py_linq_sql/sql_enumerable/sql_enumerable.py
def take_last(self, number: int) -> SQLEnumerable:
"""
Add an TAKE_LAST command to the command list.
Args:
number: Number of element to take.
Returns:
Self.
Raises:
NegativeNumberError: If number < 0.
AlterError: Indirect raise by `_check_legality`.
AlreadyExecutedError: Indirect raise by `_check_legality`.
OneError: Indirect raise by `_check_legality`.
TerminalError: Indirect raise by `_check_legality`.
"""
if number < 0:
raise NegativeNumberError()
self._check_legality(Ct.TAKE_LAST)
self.flags.limit_offset = True
self.cmd.append(Command(Ct.TAKE_LAST, DotMap(number=number)))
return self
take_while(self)
Generate a TAKE WHILE request.
Exceptions: |
|
---|
Source code in py_linq_sql/sql_enumerable/sql_enumerable.py
def take_while(self) -> None:
"""
Generate a TAKE WHILE request.
Raises:
NotImplementedError: Because it's functions is not implemented.
"""
raise NotImplementedError()
to_list(self)
Generate a TO LIST request.
Exceptions: |
|
---|
Source code in py_linq_sql/sql_enumerable/sql_enumerable.py
def to_list(self) -> None:
"""
Generate a TO LIST request.
Raises:
NotImplementedError: Because it's functions is not implemented.
"""
raise NotImplementedError()
union(self, sqle, all_=False)
Add an UNION command to the command list.
Parameters: |
|
---|
Returns: |
|
---|
Exceptions: |
|
---|
Source code in py_linq_sql/sql_enumerable/sql_enumerable.py
def union(self, sqle: SQLEnumerable, all_: bool = False) -> SQLEnumerable:
"""
Add an UNION command to the command list.
Args:
sqle: An other SQLEnumerable to make the union.
all_: Boolean to know if we want an UNION (False) or an UNION ALL (True).
By default: False.
Returns:
Self.
Raises:
AlterError: Indirect raise by `_check_legality`.
AlreadyExecutedError: Indirect raise by `_check_legality`.
DeleteError: Indirect raise by `build`.
LengthMismatchError: Indirect raise by `build`.
NeedSelectError: Indirect raise by `_check_legality`.
NeedWhereError: Indirect raise by `build`.
OneError: Indirect raise by `_check_legality`.
psycopg.Error: Indirect raise by `build`.
TableError: Indirect raise by `build`.
TerminalError: Indirect raise by `_check_legality`.
TooManyReturnValueError: Indirect raise by `build`.
TypeError: Indirect raise by `build`.
TypeOperatorError: Indirect raise by `build`.
ValueError: Indirect raise by `build`.
"""
self._check_legality(Ct.UNION)
self._check_select(sqle, "intersect")
sqle_data = SQLEnumerableData(
sqle.connection,
sqle.flags,
sqle.cmd,
sqle.table,
sqle.length,
)
built_sqle_2 = build(sqle_data)
self.flags.terminal = Terminal.UNION
self.cmd.append(Command(Ct.UNION, DotMap(built_sqle_2=built_sqle_2, all_=all_)))
return self
update(self, fquery)
Add an UPDATE command to the command list.
Parameters: |
|
---|
Returns: |
|
---|
Exceptions: |
|
---|
Source code in py_linq_sql/sql_enumerable/sql_enumerable.py
def update(self, fquery: Callable[[MagicDotPath], MagicDotPath]) -> SQLEnumerable:
"""
Add an UPDATE command to the command list.
Args:
fquery: Lambda function to get path of the modification.
Returns:
Self.
Raises:
ReadOnlyPermissionDeniedError: If we try to alter the database and the
config is in readonly.
AlterError: Indirect raise by `_check_legality`.
AlreadyExecutedError: Indirect raise by `_check_legality`.
OneError: Indirect raise by `_check_legality`.
OtherThanWhereError: Indirect raise by `_check_legality`.
ReadOnlyPermissionDeniedError: Indirect raise by `_check_legality`.
TerminalError: Indirect raise by `_check_legality`.
"""
# Can be test only with a config file and read_only = true,
# but this feature is test in:
# tests/algo/test_config.py with the test: test_readonly_true
if is_read_only(): # pragma: no cover
raise ReadOnlyPermissionDeniedError()
self._check_legality(Ct.UPDATE)
self.flags.alter = True
self.cmd.append(Command(Ct.UPDATE, DotMap(fquery=fquery)))
return self
where(self, fquery)
Add an WHERE command to the command list.
Parameters: |
|
---|
Returns: |
|
---|
Exceptions: |
|
---|
Source code in py_linq_sql/sql_enumerable/sql_enumerable.py
def where(self, fquery: LambdaMagicDotPath) -> SQLEnumerable:
"""
Add an WHERE command to the command list.
Args:
fquery: Lambda function the get the path of the predicate.
Returns:
Self.
Raises:
AlterError: Indirect raise by `_check_legality`.
AlreadyExecutedError: Indirect raise by `_check_legality`.
OneError: Indirect raise by `_check_legality`.
TerminalError: Indirect raise by `_check_legality`.
"""
self._check_legality(Ct.WHERE)
self.cmd.append(Command(Ct.WHERE, DotMap(fquery=fquery)))
return self
zip(self)
Generate a ZIP request.
Exceptions: |
|
---|
Source code in py_linq_sql/sql_enumerable/sql_enumerable.py
def zip(self) -> None:
"""
Generate a ZIP request.
Raises:
NotImplementedError: Because it's functions is not implemented.
"""
raise NotImplementedError()
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))
Functions API
Database functions
File to manage the data base, connect, get a cursor on this connection.
connect(user, password, host, port, database)
Connect to a database.
Parameters: |
|
---|
Returns: |
|
---|
Exceptions: |
|
---|
Source code in py_linq_sql/utils/db.py
def connect(
user: str,
password: str,
host: str,
port: str,
database: str,
) -> Connection:
"""
Connect to a database.
Args:
user: User used to log in to the database.
password: Password of the user.
host: Host used for the connection.
port: Port used for the connection.
database: Database to which we want to connect.
Returns:
A connection to a database if all is working.
Raises:
PSQLConnectionError: If we have an error connecting to the database.
"""
try: # pragma: no cover
connection = psycopg.connect(
user=user,
password=password,
host=host,
port=port,
dbname=database,
)
return connection
except (Exception, Error) as err: # pragma: no cover
raise PSQLConnectionError(f"Error while connecting to posgeSQL {err}") from err
get_cursor(connection)
Get a cursor on the given connection.
Parameters: |
|
---|
Returns: |
|
---|
Source code in py_linq_sql/utils/db.py
def get_cursor(connection: Connection) -> Cursor:
"""
Get a cursor on the given connection.
Args:
connection: A connection to a database on which we want to execute commands.
Returns:
A cursor on the given connection to a data base to execute commands.
"""
cursor = connection.cursor(row_factory=psycopg.rows.namedtuple_row)
return cursor
Aggregate functions
Aggregate functions used in py-linq-sql.
get_aggregate(mdpa, sqle=None)
Get all built aggregate.
Parameters: |
|
---|
Returns: |
|
---|
Exceptions: |
|
---|
Source code in py_linq_sql/utils/functions/aggregate_functions.py
def get_aggregate(
mdpa: MagicDotPathAggregate | Tuple[MagicDotPathAggregate],
sqle: SQLEnumerableData | None = None,
) -> str:
"""
Get all built aggregate.
Args:
mdpa: MagicDotPathAggregate or tuple of MagicDotPathAggregate to build.
sqle: SQLEnumerableData with connection, flags, list of commands and a table.
Returns:
All aggregate with paths as str.
Raises:
TypeError: If mdpa id not a MagicDotPathAggregate
or indirect raise by `get_one_aggregate`.
psycopg.Error: Indirect raise by `get_one_aggregate`.
TableError: Indirect raise by `get_one_aggregate`.
"""
result = []
match mdpa:
case MagicDotPathAggregate():
result.append(get_one_aggregate(mdpa, mdpa.cast_type, sqle))
case tuple():
for idx, element in enumerate(mdpa):
if not isinstance(element, MagicDotPathAggregate):
raise TypeError(
"get_aggregate_path take only MagicDotPathAggregate "
"or tuple of MagicDotPathAggregate.",
)
result.append(get_one_aggregate(element, element.cast_type, sqle))
if not idx == len(mdpa) - 1:
result.append(",")
case _:
raise TypeError(
"get_aggregate_path take only MagicDotPathAggregate "
"or tuple of MagicDotPathAggregate.",
)
return " ".join(result)
get_one_aggregate(mdpa, cast_type, sqle)
Get one built aggregate.
Parameters: |
|
---|
Returns: |
|
---|
Exceptions: |
|
---|
Source code in py_linq_sql/utils/functions/aggregate_functions.py
def get_one_aggregate(
mdpa: MagicDotPathAggregate,
cast_type: type,
sqle: SQLEnumerableData | None,
) -> str:
"""
Get one built aggregate.
Args:
mdpa: MagicDotPathAggregate to build.
cast_type: Type in which we want to cast the path(s). Its optional.
sqle: SQLEnumerableData with connection, flags, list of commands and a table.
Returns:
An aggregate with path as str.
Raises:
psycopg.Error: Indirect raise by `get_path` or `get_one_concat_aggregate`.
TableError: Indirect raise by `get_path` or `get_one_concat_aggregate`.
TypeError: Indirect raise by `get_path` or `get_one_concat_aggregate`.
"""
if mdpa.operand == AggregateType.CONCAT:
return get_one_concat_aggregate(mdpa, sqle)
if cast_type == str:
result = [f"{mdpa.operand}({get_path(mdpa.mdp, sqle)[0]})"]
else:
result = [f"{mdpa.operand}(CAST({get_path(mdpa.mdp, sqle)[0]} AS"]
casted_type = get_good_type(cast_type)
if not casted_type:
raise TypeError(
f"Group_by take only int, float, decimal or date type, not {cast_type}",
)
result.append(casted_type)
return " ".join(result)
get_one_concat_aggregate(mdpa, sqle)
Get one built concat aggregate.
Parameters: |
|
---|
Returns: |
|
---|
Exceptions: |
|
---|
Source code in py_linq_sql/utils/functions/aggregate_functions.py
def get_one_concat_aggregate(
mdpa: MagicDotPathAggregate,
sqle: SQLEnumerableData | None,
) -> str:
"""
Get one built concat aggregate.
Args:
mdpa: MagicDotPathAggregate to build.
sqle: SQLEnumerableData with connection, flags, list of commands and a table.
Returns:
A concat aggregate with path as str.
Raises:
psycopg.Error: Indirect raise by `get_path`.
TableError: Indirect raise by `get_path`.
TypeError: Indirect raise by `get_path`.
"""
return f"{mdpa.operand}({get_path(mdpa.mdp, sqle, True)[0]}, '{mdpa.separator}')"
Join functions
Join functions used in py-linq-sql.
join_get_intersect(join_type, outer_key_paths, inner_key_paths)
Get the correct WHERE subcommand depending on the type of join.
Parameters: |
|
---|
Returns: |
|
---|
Source code in py_linq_sql/utils/functions/join_functions.py
def join_get_intersect(
join_type: JoinType,
outer_key_paths: List[str],
inner_key_paths: List[str],
) -> str | None:
"""
Get the correct WHERE subcommand depending on the type of join.
Args:
join_type: type of join.
outer_key_paths: paths of the values to be compared of the outer SQLEnumerable.
inner_key_paths: paths of the values to be compared og the inner SQLEnumerable.
Return:
A WHERE subcommand with the correct syntax for join or None
if we make join with intersection.
"""
result = []
match join_type:
case JoinType.LEFT_MINUS_INTERSECT | JoinType.RIGHT_MINUS_INTERSECT:
result.append("WHERE")
result.append(
" AND ".join(
[f"{path_inner_k} IS NULL" for path_inner_k in inner_key_paths],
),
)
return " ".join(result)
case JoinType.FULL_MINUS_INTERSECT:
result.append("WHERE")
result.append(
" AND ".join(
[
f"{path_outer_k} IS NULL OR {path_inner_k} IS NULL"
for (path_outer_k, path_inner_k) in zip(
outer_key_paths,
inner_key_paths,
)
],
),
)
return " ".join(result)
case _:
return None
join_get_paths(outer, inner, inner_key, outer_key, result_function)
Get all paths for join.
Parameters: |
|
---|
Returns: |
|
---|
Exceptions: |
|
---|
Source code in py_linq_sql/utils/functions/join_functions.py
def join_get_paths(
outer: SQLEnumerableData,
inner: SQLEnumerableData,
inner_key: LambdaMagicDotPath,
outer_key: LambdaMagicDotPath,
result_function: Callable[
[BaseMagicDotPath, BaseMagicDotPath],
BaseMagicDotPath | Tuple[BaseMagicDotPath] | Dict[str, BaseMagicDotPath],
]
| None,
) -> DotMap:
"""
Get all paths for join.
Args:
outer: An other SQLEnumerable to make the join.
inner: An SQLEnumerable to make the join.
inner_key: lambda to select the value to be compared of the inner
SQLEnumerable.
outer_key: lambda to select the value to be compared of the outer
SQLEnumerable.
result_function: lambda the select the values to be returned.
Returns:
All paths we will used in the join.
Raises:
psycopg.Error: Indirect raise by `get_path`.
TableError: Indirect raise by `get_path`.
TypeError: Indirect raise by `get_path`.
"""
obj_inner = MagicDotPath(inner.connection, with_table=inner.table)
obj_outer = MagicDotPath(outer.connection, with_table=outer.table)
paths = DotMap(
select_paths=None,
outer_key_paths=get_path(outer_key(obj_outer)),
inner_key_paths=get_path(inner_key(obj_inner)),
)
if result_function:
paths.select_paths = get_path(result_function(obj_inner, obj_outer))
return paths
Path functions
Path functions used in py-linq-sql.
get_path(magic_dp, sqle=None, force=False)
Get path from a MagicDotPath.
Parameters: |
|
---|
Returns: |
|
---|
Exceptions: |
|
---|
Source code in py_linq_sql/utils/functions/path_functions.py
def get_path(
magic_dp: BaseMagicDotPath | Tuple[BaseMagicDotPath] | Dict[str, BaseMagicDotPath],
sqle: SQLEnumerableData | None = None,
force: bool = False,
) -> List[str]:
"""
Get path from a MagicDotPath.
Args:
magic_dp: A MagicDotPath objects contains a bases with element of the future
path.
sqle: SQLEnumerableData with connection, flags, list of commands and a table.
force: True if we want to force the json path in text, False otherwise.
By default: False.
Returns:
List a path transform by jsonb_path.
Raises:
TypeError: If magic_dp is not a subclass of BaseMagicDotPath.
psycopg.Error: Indirect raise by `_get_path_base_mdp`.
TableError: Indirect raise by `_get_path_base_mdp`.
"""
paths: List[str] = []
match magic_dp:
case BaseMagicDotPath():
_get_path_base_mdp(magic_dp, sqle, force, paths)
case tuple():
for element in magic_dp:
_get_path_base_mdp(element, sqle, force, paths)
case dict():
for element in magic_dp.values():
_get_path_base_mdp(element, sqle, force, paths)
case _:
raise TypeError(
"`get_path()` take only BaseMagicDotPath or tuple of BaseMagicDotPath.",
)
return paths
get_paths(fquery, sqle, as_str=False)
Get jsonb paths to build commands.
The format of paths give by the function: person->'address'->'zip_code'
Parameters: |
|
---|
Returns: |
|
---|
Exceptions: |
|
---|
Source code in py_linq_sql/utils/functions/path_functions.py
def get_paths(
fquery: LambdaMagicDotPath,
sqle: SQLEnumerableData,
as_str: bool = False,
) -> List[str]:
"""
Get jsonb paths to build commands.
The format of paths give by the function: person->'address'->'zip_code'
Args:
fquery: Lambda function to get the path(s)
sqle: SQLEnumerableData with connection, flags, list of commands and a table.
as_str: False if we want basic paths, True if we want force the paths on string.
Returns:
List of paths.
Raises:
psycopg.Error: Indirect raise by `get_path`.
TableError: Indirect raise by `get_path`.
TypeError: Indirect raise by `get_path`.
TypeOperatorError: Indirect raise by `BaseMagicDotPath._get_number_operator`
or `BaseMagicDotPath._get_generic_operator`.
"""
return get_path(fquery(MagicDotPath(sqle.connection)), sqle, as_str)
get_update_path(path)
Get the correct format of path for UPDATE.
We need a path like: {0, 1, 2, 3}
.
Parameters: |
|
---|
Returns: |
|
---|
Source code in py_linq_sql/utils/functions/path_functions.py
def get_update_path(path: str) -> str:
"""
Get the correct format of path for UPDATE.
We need a path like: `{0, 1, 2, 3}`.
Args:
path: The path to get in the correct format for UPDATE.
Returns:
The path in the correct format for UPDATE.
"""
tmp = re.split(r">|-|'", path)
# Delete the first, is the column
tmp = tmp[1:]
# Join all not None str in tmp with the function filter()
# (https://docs.python.org/3/library/functions.html#filter)
result = ",".join(filter(None, tmp))
return "{" + result + "}"
Predicate functions
Predicate functions used in py-linq-sql.
get_one_predicate_as_str(sqle, mdpo)
Get one predicate as string with the correct cast type and the correct prefix.
Parameters: |
|
---|
Returns: |
|
---|
Exceptions: |
|
---|
Source code in py_linq_sql/utils/functions/predicate_functions.py
def get_one_predicate_as_str(
sqle: SQLEnumerableData,
mdpo: BaseMagicDotPath,
) -> str:
"""
Get one predicate as string with the correct cast type and the correct prefix.
Args:
sqle: SQLEnumerable with connection, flags, list of commands and a table.
mdpo: BasemagicDotPath to build the predicate.
Returns:
A predicate as a string with the correct cast type and prefix.
Raises:
psycopg.Error: Indirect raise by `get_path`.
TableError: Indirect raise by `get_path`.
TypeError: Indirect raise by `get_path`.
"""
return get_path(mdpo, sqle)[0]
get_predicates_as_str(result, fquery, sqle)
Get all predicates as string with the correct cast type and the correct prefix.
Parameters: |
|
---|
Returns: |
|
---|
Exceptions: |
|
---|
Source code in py_linq_sql/utils/functions/predicate_functions.py
def get_predicates_as_str(
result: List[str],
fquery: LambdaMagicDotPath,
sqle: SQLEnumerableData,
) -> None:
"""
Get all predicates as string with the correct cast type and the correct prefix.
Args:
result: List contains the request.
fquery: Lambda function to get paths.
sqle: SQLEnumerableData with connection, flags, list of commands and a table.
Returns:
Predicates as a string with the correct cast type and prefix.
Raises:
TypeError: If the type of mdp_w_path isn't BaseMagicDotPath
or tuple of BaseMagicDotPath,
or indirect raise by `get_one_predicate_as_str`.
psycopg.Error: Indirect raise by `get_one_predicate_as_str`.
TableError: Indirect raise by `get_one_predicate_as_str`.
TypeOperatorError: Indirect raise by `BaseMagicDotPath._get_number_operator`
or `BaseMagicDotPath._get_generic_operator`.
"""
mdp_w_path = fquery(MagicDotPath(sqle.connection))
match mdp_w_path:
case BaseMagicDotPath():
result.append(get_one_predicate_as_str(sqle, mdp_w_path))
case tuple():
result.append(
" AND ".join(
[get_one_predicate_as_str(sqle, mdp) for mdp in mdp_w_path],
),
)
case _:
raise TypeError(
"Only BaseMagicDotPath or tuple of BaseMagicDotPath are accepted.",
)
MagicDotPath mathematical functions
All mathematics functions for MagicDotPath.
cbrt(mdp)
Cube root function for a MagicDotPath.
From psql docs: Cube root.
See: https://www.postgresql.org/docs/current/functions-math.html
Parameters: |
|
---|
Returns: |
|
---|
Source code in py_linq_sql/utils/functions/magic_dp_maths_functions.py
def cbrt(mdp: BaseMagicDotPath) -> MagicDotPathWithOp:
"""
Cube root function for a MagicDotPath.
From psql docs: Cube root.
See: https://www.postgresql.org/docs/current/functions-math.html
Args:
mdp: MagicDotPath on which we apply the cube root.
Returns:
MagicDotPathWithOp with one operand and on the correct operator.
"""
return mdp._get_one_operand_operator( # pylint: disable=protected-access
MathFunctType.CBRT,
)
ceil(mdp)
Ceil function for a MagicDotPath.
From psql docs: Nearest integer greater than or equal to argument.
See: https://www.postgresql.org/docs/current/functions-math.html
Parameters: |
|
---|
Returns: |
|
---|
Source code in py_linq_sql/utils/functions/magic_dp_maths_functions.py
def ceil(mdp: BaseMagicDotPath) -> MagicDotPathWithOp:
"""
Ceil function for a MagicDotPath.
From psql docs: Nearest integer greater than or equal to argument.
See: https://www.postgresql.org/docs/current/functions-math.html
Args:
mdp: MagicDotPath on which we apply the ceil.
Returns:
MagicDotPathWithOp with one operand and on the correct operator.
"""
return mdp._get_one_operand_operator( # pylint: disable=protected-access
MathFunctType.CEIL,
)
degrees(mdp)
Degrees function for a MagicDotPath.
From psql docs: Converts radians to degrees.
See: https://www.postgresql.org/docs/current/functions-math.html
Parameters: |
|
---|
Returns: |
|
---|
Source code in py_linq_sql/utils/functions/magic_dp_maths_functions.py
def degrees(mdp: BaseMagicDotPath) -> MagicDotPathWithOp:
"""
Degrees function for a MagicDotPath.
From psql docs: Converts radians to degrees.
See: https://www.postgresql.org/docs/current/functions-math.html
Args:
mdp: MagicDotPath on which we apply the degrees.
Returns:
MagicDotPathWithOp with one operand and on the correct operator.
"""
return mdp._get_one_operand_operator( # pylint: disable=protected-access
MathFunctType.DEGREES,
)
exp(mdp)
Exp function for a MagicDotPath.
From psql docs: Exponential (e raised to the given power).
See: https://www.postgresql.org/docs/current/functions-math.html
Parameters: |
|
---|
Returns: |
|
---|
Source code in py_linq_sql/utils/functions/magic_dp_maths_functions.py
def exp(mdp: BaseMagicDotPath) -> MagicDotPathWithOp:
"""
Exp function for a MagicDotPath.
From psql docs: Exponential (e raised to the given power).
See: https://www.postgresql.org/docs/current/functions-math.html
Args:
mdp: MagicDotPath on which we apply the exp.
Returns:
MagicDotPathWithOp with one operand and on the correct operator.
"""
return mdp._get_one_operand_operator( # pylint: disable=protected-access
MathFunctType.EXP,
)
factorial(mdp)
Factorial function for a MagicDotPath.
From psql docs: Factorial.
See: https://www.postgresql.org/docs/current/functions-math.html
Parameters: |
|
---|
Returns: |
|
---|
Source code in py_linq_sql/utils/functions/magic_dp_maths_functions.py
def factorial(mdp: BaseMagicDotPath) -> MagicDotPathWithOp:
"""
Factorial function for a MagicDotPath.
From psql docs: Factorial.
See: https://www.postgresql.org/docs/current/functions-math.html
Args:
mdp: MagicDotPath on which we apply the factorial.
Returns:
MagicDotPathWithOp with one operand and on the correct operator.
"""
return mdp._get_one_operand_operator( # pylint: disable=protected-access
MathFunctType.FACTORIAL,
)
floor(mdp)
Floor function for a MagicDotPath.
From psql docs: Nearest integer less than or equal to argument.
See: https://www.postgresql.org/docs/current/functions-math.html
Parameters: |
|
---|
Returns: |
|
---|
Source code in py_linq_sql/utils/functions/magic_dp_maths_functions.py
def floor(mdp: BaseMagicDotPath) -> MagicDotPathWithOp:
"""
Floor function for a MagicDotPath.
From psql docs: Nearest integer less than or equal to argument.
See: https://www.postgresql.org/docs/current/functions-math.html
Args:
mdp: MagicDotPath on which we apply the floor.
Returns:
MagicDotPathWithOp with one operand and on the correct operator.
"""
return mdp._get_one_operand_operator( # pylint: disable=protected-access
MathFunctType.FLOOR,
)
gcd(mdp, other)
Gcd function for a MagicDotPath.
From psql docs: Greatest common divisor (the largest positive number that divides both inputs with no remainder); returns 0 if both inputs are zero; available for integer, bigint, and numeric.
See: https://www.postgresql.org/docs/current/functions-math.html
Parameters: |
|
---|
Returns: |
|
---|
Exceptions: |
|
---|
Source code in py_linq_sql/utils/functions/magic_dp_maths_functions.py
def gcd(mdp: BaseMagicDotPath, other: _NUMBER_TYPE) -> MagicDotPathWithOp:
"""
Gcd function for a MagicDotPath.
From psql docs: Greatest common divisor (the largest positive number
that divides both inputs with no remainder); returns 0 if both inputs are zero;
available for integer, bigint, and numeric.
See: https://www.postgresql.org/docs/current/functions-math.html
Args:
mdp: MagicDotPath on which we apply the gcd.
other: An other element for the comparison.
Returns:
MagicDotPathWithOp with 2 operand and on the correct operator.
Raises:
TypeOperatorError: Indirect raise by `BaseMagicDotPath._get_number_operator`
"""
return mdp._get_number_operator( # pylint: disable=protected-access
other,
MathFunctType.GCD,
)
greatest(mdp, other)
Greatest function for a MagicDotPath.
This is equivalent to the max
function in python.
From psql docs: The GREATEST function select the largest value from a list of any number of expressions. The expressions must all be convertible to a common data type, which will be the type of the result
See: https://www.postgresql.org/docs/current/functions-conditional.html#FUNCTIONS-GREATEST-LEAST
Parameters: |
|
---|
Returns: |
|
---|
Exceptions: |
|
---|
Source code in py_linq_sql/utils/functions/magic_dp_maths_functions.py
def greatest(
mdp: BaseMagicDotPath, other: _NUMBER_TYPE | BaseMagicDotPath
) -> MagicDotPathWithOp:
"""
Greatest function for a MagicDotPath.
This is equivalent to the `max` function in python.
From psql docs: The GREATEST function select the largest value from a list of any
number of expressions. The expressions must all be convertible to a common data
type, which will be the type of the result
See:
https://www.postgresql.org/docs/current/functions-conditional.html#FUNCTIONS-GREATEST-LEAST
Args:
mdp: MagicDotPath on which we apply the greatest.
other: An other element for
the comparison.
Returns:
MagicDotPathWithOp with 2 operand and on the correct operator.
Raises:
TypeOperatorError: Indirect raise by `BaseMagicDotPath._get_number_operator`
"""
return mdp._get_number_operator( # pylint: disable=protected-access
other,
MathFunctType.GREATEST,
)
lcm(mdp, other)
Lcm function for a MagicDotPath.
From psql docs: Least common multiple (the smallest strictly positive number that is an integral multiple of both inputs); returns 0 if either input is zero; available for integer, bigint, and numeric.
See: https://www.postgresql.org/docs/current/functions-math.html
Parameters: |
|
---|
Returns: |
|
---|
Exceptions: |
|
---|
Source code in py_linq_sql/utils/functions/magic_dp_maths_functions.py
def lcm(mdp: BaseMagicDotPath, other: _NUMBER_TYPE) -> MagicDotPathWithOp:
"""
Lcm function for a MagicDotPath.
From psql docs: Least common multiple (the smallest strictly positive number
that is an integral multiple of both inputs); returns 0 if either input is zero;
available for integer, bigint, and numeric.
See: https://www.postgresql.org/docs/current/functions-math.html
Args:
mdp: MagicDotPath on which we apply the lcm.
other: An other element for the comparison.
Returns:
MagicDotPathWithOp with 2 operand and on the correct operator.
Raises:
TypeOperatorError: Indirect raise by `BaseMagicDotPath._get_number_operator`
"""
return mdp._get_number_operator( # pylint: disable=protected-access
other,
MathFunctType.LCM,
)
least(mdp, other)
Greatest function for a MagicDotPath.
This is equivalent to the min
function in python.
From psql docs: The LEAST function select the smallest value from a list of any number of expressions. The expressions must all be convertible to a common data type, which will be the type of the result
See: https://www.postgresql.org/docs/current/functions-conditional.html#FUNCTIONS-GREATEST-LEAST
Parameters: |
|
---|
Returns: |
|
---|
Exceptions: |
|
---|
Source code in py_linq_sql/utils/functions/magic_dp_maths_functions.py
def least(
mdp: BaseMagicDotPath, other: _NUMBER_TYPE | BaseMagicDotPath
) -> MagicDotPathWithOp:
"""
Greatest function for a MagicDotPath.
This is equivalent to the `min` function in python.
From psql docs: The LEAST function select the smallest value from a list of any
number of expressions. The expressions must all be convertible to a common data
type, which will be the type of the result
See:
https://www.postgresql.org/docs/current/functions-conditional.html#FUNCTIONS-GREATEST-LEAST
Args:
mdp: MagicDotPath on which we apply the least.
other: An other element for
the comparison.
Returns:
MagicDotPathWithOp with 2 operand and on the correct operator.
Raises:
TypeOperatorError: Indirect raise by `BaseMagicDotPath._get_number_operator`
"""
return mdp._get_number_operator( # pylint: disable=protected-access
other,
MathFunctType.LEAST,
)
ln(mdp)
Ln function for a MagicDotPath.
From psql docs: Natural logarithm.
See: https://www.postgresql.org/docs/current/functions-math.html
Parameters: |
|
---|
Returns: |
|
---|
Source code in py_linq_sql/utils/functions/magic_dp_maths_functions.py
def ln(
mdp: BaseMagicDotPath,
) -> MagicDotPathWithOp:
"""
Ln function for a MagicDotPath.
From psql docs: Natural logarithm.
See: https://www.postgresql.org/docs/current/functions-math.html
Args:
mdp: MagicDotPath on which we apply the ln.
Returns:
MagicDotPathWithOp with one operand and on the correct operator.
"""
return mdp._get_one_operand_operator( # pylint: disable=protected-access
MathFunctType.LN,
)
log(mdp, other)
Log function for a MagicDotPath.
From psql docs: Logarithm of x to base b.
See: https://www.postgresql.org/docs/current/functions-math.html
Parameters: |
|
---|
Returns: |
|
---|
Exceptions: |
|
---|
Source code in py_linq_sql/utils/functions/magic_dp_maths_functions.py
def log(mdp: BaseMagicDotPath, other: _NUMBER_TYPE) -> MagicDotPathWithOp:
"""
Log function for a MagicDotPath.
From psql docs: Logarithm of x to base b.
See: https://www.postgresql.org/docs/current/functions-math.html
Args:
mdp: MagicDotPath on which we apply the log.
other: Base for the logarithm.
Returns:
MagicDotPathWithOp with one operand and on the correct operator.
Raises:
TypeOperatorError: Indirect raise by `BaseMagicDotPath._get_number_operator`
"""
return mdp._get_number_operator( # pylint: disable=protected-access
other,
MathFunctType.LOG,
)
log10(mdp)
Log10 function for a MagicDotPath.
From psql docs: Base 10 logarithm.
See: https://www.postgresql.org/docs/current/functions-math.html
Parameters: |
|
---|
Returns: |
|
---|
Source code in py_linq_sql/utils/functions/magic_dp_maths_functions.py
def log10(mdp: BaseMagicDotPath) -> MagicDotPathWithOp:
"""
Log10 function for a MagicDotPath.
From psql docs: Base 10 logarithm.
See: https://www.postgresql.org/docs/current/functions-math.html
Args:
mdp: MagicDotPath on which we apply the log10.
Returns:
MagicDotPathWithOp with one operand and on the correct operator.
"""
return mdp._get_one_operand_operator( # pylint: disable=protected-access
MathFunctType.LOG10,
)
min_scale(mdp)
Min scale function for a MagicDotPath.
From psql docs: Minimum scale (number of fractional decimal digits) needed to represent the supplied value precisely.
See: https://www.postgresql.org/docs/current/functions-math.html
Parameters: |
|
---|
Returns: |
|
---|
Source code in py_linq_sql/utils/functions/magic_dp_maths_functions.py
def min_scale(mdp: BaseMagicDotPath) -> MagicDotPathWithOp:
"""
Min scale function for a MagicDotPath.
From psql docs: Minimum scale (number of fractional decimal digits) needed
to represent the supplied value precisely.
See: https://www.postgresql.org/docs/current/functions-math.html
Args:
mdp: MagicDotPath on which we apply the min scale.
Returns:
MagicDotPathWithOp with one operand and on the correct operator.
"""
return mdp._get_one_operand_operator( # pylint: disable=protected-access
MathFunctType.MIN_SCALE,
)
radians(mdp)
Radiant function for a MagicDotPath.
From psql docs: Converts degrees to radians.
See: https://www.postgresql.org/docs/current/functions-math.html
Parameters: |
|
---|
Returns: |
|
---|
Source code in py_linq_sql/utils/functions/magic_dp_maths_functions.py
def radians(mdp: BaseMagicDotPath) -> MagicDotPathWithOp:
"""
Radiant function for a MagicDotPath.
From psql docs: Converts degrees to radians.
See: https://www.postgresql.org/docs/current/functions-math.html
Args:
mdp: MagicDotPath on which we apply the radiant.
Returns:
MagicDotPathWithOp with one operand and on the correct operator.
"""
return mdp._get_one_operand_operator( # pylint: disable=protected-access
MathFunctType.RADIANS,
)
round(mdp, other=None)
Round function for a MagicDotPath.
From psql docs:
- round(mdp): Rounds to nearest integer. For numeric, ties are broken by rounding away from zero. For double precision, the tie-breaking behavior is platform dependent, but “round to nearest even” is the most common rule.
- round(mdp, other): Rounds v to s decimal places. Ties are broken by rounding away from zero.
See: https://www.postgresql.org/docs/current/functions-math.html
Parameters: |
|
---|
Returns: |
|
---|
Exceptions: |
|
---|
Source code in py_linq_sql/utils/functions/magic_dp_maths_functions.py
def round( # noqa: A001
mdp: BaseMagicDotPath,
other: _NUMBER_TYPE | None = None,
) -> MagicDotPathWithOp:
"""
Round function for a MagicDotPath.
From psql docs:
- round(mdp): Rounds to nearest integer. For numeric,
ties are broken by rounding away from zero.
For double precision, the tie-breaking behavior is platform dependent,
but “round to nearest even” is the most common rule.
- round(mdp, other): Rounds v to s decimal places.
Ties are broken by rounding away from zero.
See: https://www.postgresql.org/docs/current/functions-math.html
Args:
mdp: MagicDotPath on which we apply the round.
other: Number of decimal we want to keep.
Returns:
MagicDotPathWithOp with 2 operand and on the correct operator.
Raises:
TypeOperatorError: Indirect raise by `BaseMagicDotPath._get_number_operator`
"""
if not other:
other = 0
return mdp._get_number_operator( # pylint: disable=protected-access
other,
MathFunctType.ROUND,
)
scale(mdp)
Scale function for a MagicDotPath.
From psql docs: Scale of the argument (the number of decimal digits in the fractional part).
See: https://www.postgresql.org/docs/current/functions-math.html
Parameters: |
|
---|
Returns: |
|
---|
Source code in py_linq_sql/utils/functions/magic_dp_maths_functions.py
def scale(mdp: BaseMagicDotPath) -> MagicDotPathWithOp:
"""
Scale function for a MagicDotPath.
From psql docs: Scale of the argument
(the number of decimal digits in the fractional part).
See: https://www.postgresql.org/docs/current/functions-math.html
Args:
mdp: MagicDotPath on which we apply the scale.
Returns:
MagicDotPathWithOp with one operand and on the correct operator.
"""
return mdp._get_one_operand_operator( # pylint: disable=protected-access
MathFunctType.SCALE,
)
sign(mdp)
Sign function for a MagicDotPath.
From psql docs: Sign of the argument (-1, 0, or +1).
See: https://www.postgresql.org/docs/current/functions-math.html
Parameters: |
|
---|
Returns: |
|
---|
Source code in py_linq_sql/utils/functions/magic_dp_maths_functions.py
def sign(mdp: BaseMagicDotPath) -> MagicDotPathWithOp:
"""
Sign function for a MagicDotPath.
From psql docs: Sign of the argument (-1, 0, or +1).
See: https://www.postgresql.org/docs/current/functions-math.html
Args:
mdp: MagicDotPath on which we apply the sign.
Returns:
MagicDotPathWithOp with one operand and on the correct operator.
"""
return mdp._get_one_operand_operator( # pylint: disable=protected-access
MathFunctType.SIGN,
)
sqrt(mdp)
Squart root function for a MagicDotPath.
From psql docs: Square root.
See: https://www.postgresql.org/docs/current/functions-math.html
Parameters: |
|
---|
Returns: |
|
---|
Source code in py_linq_sql/utils/functions/magic_dp_maths_functions.py
def sqrt(mdp: BaseMagicDotPath) -> MagicDotPathWithOp:
"""
Squart root function for a MagicDotPath.
From psql docs: Square root.
See: https://www.postgresql.org/docs/current/functions-math.html
Args:
mdp: MagicDotPath on which we apply the squart root.
Returns:
MagicDotPathWithOp with one operand and on the correct operator.
"""
return mdp._get_one_operand_operator( # pylint: disable=protected-access
MathFunctType.SQRT,
)
trim_scale(mdp)
Trim Scale function for a MagicDotPath.
From psql docs: Reduces the value's scale (number of fractional decimal digits) by removing trailing zeroes.
See: https://www.postgresql.org/docs/current/functions-math.html
Parameters: |
|
---|
Returns: |
|
---|
Source code in py_linq_sql/utils/functions/magic_dp_maths_functions.py
def trim_scale(mdp: BaseMagicDotPath) -> MagicDotPathWithOp:
"""
Trim Scale function for a MagicDotPath.
From psql docs: Reduces the value's scale (number of fractional decimal digits)
by removing trailing zeroes.
See: https://www.postgresql.org/docs/current/functions-math.html
Args:
mdp: MagicDotPath on which we apply the trim scale.
Returns:
MagicDotPathWithOp with one operand and on the correct operator.
"""
return mdp._get_one_operand_operator( # pylint: disable=protected-access
MathFunctType.TRIM_SCALE,
)
trunc(mdp, other=None)
Trunc function for a MagicDotPath.
From psql docs:
- trunc(mdp): Truncates to integer (towards zero).
- trunc(mdp, other): Truncates v to s decimal places.
See: https://www.postgresql.org/docs/current/functions-math.html
Parameters: |
|
---|
Returns: |
|
---|
Exceptions: |
|
---|
Source code in py_linq_sql/utils/functions/magic_dp_maths_functions.py
def trunc(
mdp: BaseMagicDotPath,
other: _NUMBER_TYPE | None = None,
) -> MagicDotPathWithOp:
"""
Trunc function for a MagicDotPath.
From psql docs:
- trunc(mdp): Truncates to integer (towards zero).
- trunc(mdp, other): Truncates v to s decimal places.
See: https://www.postgresql.org/docs/current/functions-math.html
Args:
mdp: MagicDotPath on which we apply the trunc.
other: Number of decimal we want to keep.
Returns:
MagicDotPathWithOp with 2 operand and on the correct operator.
Raises:
TypeOperatorError: Indirect raise by `BaseMagicDotPath._get_number_operator`
"""
if not other:
other = 0
return mdp._get_number_operator( # pylint: disable=protected-access
other,
MathFunctType.TRUNC,
)
MagicDotPath trigonometric functions
All trigonometric functions for MagicDotPath.
acos(mdp)
Acos function for a MagicDotPath.
From psql docs: Inverse cosine, result in radians.
See: https://www.postgresql.org/docs/current/functions-math.html
Parameters: |
|
---|
Returns: |
|
---|
Source code in py_linq_sql/utils/functions/magic_dp_trigo_functions.py
def acos(mdp: BaseMagicDotPath) -> MagicDotPathWithOp:
"""
Acos function for a MagicDotPath.
From psql docs: Inverse cosine, result in radians.
See: https://www.postgresql.org/docs/current/functions-math.html
Args:
mdp: MagicDotPath on which we apply the acos.
Returns:
MagicDotPathWithOp with one operand and on the correct operator.
"""
return mdp._get_one_operand_operator( # pylint: disable=protected-access
TrigoFunctType.ACOS,
)
acosd(mdp)
Acosd function for a MagicDotPath.
From psql docs: Inverse cosine, result in degrees.
See: https://www.postgresql.org/docs/current/functions-math.html
Parameters: |
|
---|
Returns: |
|
---|
Source code in py_linq_sql/utils/functions/magic_dp_trigo_functions.py
def acosd(mdp: BaseMagicDotPath) -> MagicDotPathWithOp:
"""
Acosd function for a MagicDotPath.
From psql docs: Inverse cosine, result in degrees.
See: https://www.postgresql.org/docs/current/functions-math.html
Args:
mdp: MagicDotPath on which we apply the acosd.
Returns:
MagicDotPathWithOp with one operand and on the correct operator.
"""
return mdp._get_one_operand_operator( # pylint: disable=protected-access
TrigoFunctType.ACOSD,
)
asin(mdp)
Asin function for a MagicDotPath.
From psql docs: Inverse sine, result in radians.
See: https://www.postgresql.org/docs/current/functions-math.html
Parameters: |
|
---|
Returns: |
|
---|
Source code in py_linq_sql/utils/functions/magic_dp_trigo_functions.py
def asin(mdp: BaseMagicDotPath) -> MagicDotPathWithOp:
"""
Asin function for a MagicDotPath.
From psql docs: Inverse sine, result in radians.
See: https://www.postgresql.org/docs/current/functions-math.html
Args:
mdp: MagicDotPath on which we apply the asin.
Returns:
MagicDotPathWithOp with one operand and on the correct operator.
"""
return mdp._get_one_operand_operator( # pylint: disable=protected-access
TrigoFunctType.ASIN,
)
asind(mdp)
Asind function for a MagicDotPath.
From psql docs: Inverse sine, result in degrees.
See: https://www.postgresql.org/docs/current/functions-math.html
Parameters: |
|
---|
Returns: |
|
---|
Source code in py_linq_sql/utils/functions/magic_dp_trigo_functions.py
def asind(mdp: BaseMagicDotPath) -> MagicDotPathWithOp:
"""
Asind function for a MagicDotPath.
From psql docs: Inverse sine, result in degrees.
See: https://www.postgresql.org/docs/current/functions-math.html
Args:
mdp: MagicDotPath on which we apply the asind.
Returns:
MagicDotPathWithOp with one operand and on the correct operator.
"""
return mdp._get_one_operand_operator( # pylint: disable=protected-access
TrigoFunctType.ASIND,
)
atan(mdp)
Atan function for a MagicDotPath.
From psql docs: Inverse tangent, result in radians.
See: https://www.postgresql.org/docs/current/functions-math.html
Parameters: |
|
---|
Returns: |
|
---|
Source code in py_linq_sql/utils/functions/magic_dp_trigo_functions.py
def atan(mdp: BaseMagicDotPath) -> MagicDotPathWithOp:
"""
Atan function for a MagicDotPath.
From psql docs: Inverse tangent, result in radians.
See: https://www.postgresql.org/docs/current/functions-math.html
Args:
mdp: MagicDotPath on which we apply the atan.
Returns:
MagicDotPathWithOp with one operand and on the correct operator.
"""
return mdp._get_one_operand_operator( # pylint: disable=protected-access
TrigoFunctType.ATAN,
)
atan2(mdp, other)
Atan2 function for a MagicDotPath.
From psql docs: Inverse tangent of y/x, result in radians.
See: https://www.postgresql.org/docs/current/functions-math.html
Parameters: |
|
---|
Returns: |
|
---|
Exceptions: |
|
---|
Source code in py_linq_sql/utils/functions/magic_dp_trigo_functions.py
def atan2(mdp: BaseMagicDotPath, other: _NUMBER_TYPE) -> MagicDotPathWithOp:
"""
Atan2 function for a MagicDotPath.
From psql docs: Inverse tangent of y/x, result in radians.
See: https://www.postgresql.org/docs/current/functions-math.html
Args:
mdp: MagicDotPath representing x.
other: number representing y.
Returns:
MagicDotPathWithOp with one operand and on the correct operator.
Raises:
TypeOperatorError: Indirect raise by `BaseMagicDotPath._get_number_operator`
"""
return mdp._get_number_operator( # pylint: disable=protected-access
other,
TrigoFunctType.ATAN2,
)
atan2d(mdp, other)
Atan2d function for a MagicDotPath.
From psql docs: Inverse tangent of y/x, result in degrees.
See: https://www.postgresql.org/docs/current/functions-math.html
Parameters: |
|
---|
Returns: |
|
---|
Exceptions: |
|
---|
Source code in py_linq_sql/utils/functions/magic_dp_trigo_functions.py
def atan2d(mdp: BaseMagicDotPath, other: _NUMBER_TYPE) -> MagicDotPathWithOp:
"""
Atan2d function for a MagicDotPath.
From psql docs: Inverse tangent of y/x, result in degrees.
See: https://www.postgresql.org/docs/current/functions-math.html
Args:
mdp: MagicDotPath representing x.
other: number representing y.
Returns:
MagicDotPathWithOp with one operand and on the correct operator.
Raises:
TypeOperatorError: Indirect raise by `BaseMagicDotPath._get_number_operator`
"""
return mdp._get_number_operator( # pylint: disable=protected-access
other,
TrigoFunctType.ATAN2D,
)
atand(mdp)
Atand function for a MagicDotPath.
From psql docs: Inverse tangent, result in degrees.
See: https://www.postgresql.org/docs/current/functions-math.html
Parameters: |
|
---|
Returns: |
|
---|
Source code in py_linq_sql/utils/functions/magic_dp_trigo_functions.py
def atand(mdp: BaseMagicDotPath) -> MagicDotPathWithOp:
"""
Atand function for a MagicDotPath.
From psql docs: Inverse tangent, result in degrees.
See: https://www.postgresql.org/docs/current/functions-math.html
Args:
mdp: MagicDotPath on which we apply the atand.
Returns:
MagicDotPathWithOp with one operand and on the correct operator.
"""
return mdp._get_one_operand_operator( # pylint: disable=protected-access
TrigoFunctType.ATAND,
)
cos(mdp)
Cos function for a MagicDotPath.
From psql docs: Cosine, argument in radians.
See: https://www.postgresql.org/docs/current/functions-math.html
Parameters: |
|
---|
Returns: |
|
---|
Source code in py_linq_sql/utils/functions/magic_dp_trigo_functions.py
def cos(mdp: BaseMagicDotPath) -> MagicDotPathWithOp:
"""
Cos function for a MagicDotPath.
From psql docs: Cosine, argument in radians.
See: https://www.postgresql.org/docs/current/functions-math.html
Args:
mdp: MagicDotPath on which we apply the cos.
Returns:
MagicDotPathWithOp with one operand and on the correct operator.
"""
return mdp._get_one_operand_operator( # pylint: disable=protected-access
TrigoFunctType.COS,
)
cosd(mdp)
Cosd function for a MagicDotPath.
From psql docs: Cosine, argument in degrees.
See: https://www.postgresql.org/docs/current/functions-math.html
Parameters: |
|
---|
Returns: |
|
---|
Source code in py_linq_sql/utils/functions/magic_dp_trigo_functions.py
def cosd(mdp: BaseMagicDotPath) -> MagicDotPathWithOp:
"""
Cosd function for a MagicDotPath.
From psql docs: Cosine, argument in degrees.
See: https://www.postgresql.org/docs/current/functions-math.html
Args:
mdp: MagicDotPath on which we apply the cosd.
Returns:
MagicDotPathWithOp with one operand and on the correct operator.
"""
return mdp._get_one_operand_operator( # pylint: disable=protected-access
TrigoFunctType.COSD,
)
cot(mdp)
Cot function for a MagicDotPath.
From psql docs: Cotangent, argument in radians.
See: https://www.postgresql.org/docs/current/functions-math.html
Parameters: |
|
---|
Returns: |
|
---|
Source code in py_linq_sql/utils/functions/magic_dp_trigo_functions.py
def cot(mdp: BaseMagicDotPath) -> MagicDotPathWithOp:
"""
Cot function for a MagicDotPath.
From psql docs: Cotangent, argument in radians.
See: https://www.postgresql.org/docs/current/functions-math.html
Args:
mdp: MagicDotPath on which we apply the cot.
Returns:
MagicDotPathWithOp with one operand and on the correct operator.
"""
return mdp._get_one_operand_operator( # pylint: disable=protected-access
TrigoFunctType.COT,
)
cotd(mdp)
Cotd function for a MagicDotPath.
From psql docs: Cotangent, argument in degrees.
See: https://www.postgresql.org/docs/current/functions-math.html
Parameters: |
|
---|
Returns: |
|
---|
Source code in py_linq_sql/utils/functions/magic_dp_trigo_functions.py
def cotd(mdp: BaseMagicDotPath) -> MagicDotPathWithOp:
"""
Cotd function for a MagicDotPath.
From psql docs: Cotangent, argument in degrees.
See: https://www.postgresql.org/docs/current/functions-math.html
Args:
mdp: MagicDotPath on which we apply the cotd.
Returns:
MagicDotPathWithOp with one operand and on the correct operator.
"""
return mdp._get_one_operand_operator( # pylint: disable=protected-access
TrigoFunctType.COTD,
)
sin(mdp)
Sin function for a MagicDotPath.
From psql docs: Sine, argument in radians.
See: https://www.postgresql.org/docs/current/functions-math.html
Parameters: |
|
---|
Returns: |
|
---|
Source code in py_linq_sql/utils/functions/magic_dp_trigo_functions.py
def sin(mdp: BaseMagicDotPath) -> MagicDotPathWithOp:
"""
Sin function for a MagicDotPath.
From psql docs: Sine, argument in radians.
See: https://www.postgresql.org/docs/current/functions-math.html
Args:
mdp: MagicDotPath on which we apply the sin.
Returns:
MagicDotPathWithOp with one operand and on the correct operator.
"""
return mdp._get_one_operand_operator( # pylint: disable=protected-access
TrigoFunctType.SIN,
)
sind(mdp)
Sind function for a MagicDotPath.
From psql docs: Sine, argument in degrees.
See: https://www.postgresql.org/docs/current/functions-math.html
Parameters: |
|
---|
Returns: |
|
---|
Source code in py_linq_sql/utils/functions/magic_dp_trigo_functions.py
def sind(mdp: BaseMagicDotPath) -> MagicDotPathWithOp:
"""
Sind function for a MagicDotPath.
From psql docs: Sine, argument in degrees.
See: https://www.postgresql.org/docs/current/functions-math.html
Args:
mdp: MagicDotPath on which we apply the sind.
Returns:
MagicDotPathWithOp with one operand and on the correct operator.
"""
return mdp._get_one_operand_operator( # pylint: disable=protected-access
TrigoFunctType.SIND,
)
tan(mdp)
Tand function for a MagicDotPath.
From psql docs: Tangent, argument in radians.
See: https://www.postgresql.org/docs/current/functions-math.html
Parameters: |
|
---|
Returns: |
|
---|
Source code in py_linq_sql/utils/functions/magic_dp_trigo_functions.py
def tan(mdp: BaseMagicDotPath) -> MagicDotPathWithOp:
"""
Tand function for a MagicDotPath.
From psql docs: Tangent, argument in radians.
See: https://www.postgresql.org/docs/current/functions-math.html
Args:
mdp: MagicDotPath on which we apply the tan.
Returns:
MagicDotPathWithOp with one operand and on the correct operator.
"""
return mdp._get_one_operand_operator( # pylint: disable=protected-access
TrigoFunctType.TAN,
)
tand(mdp)
Tand function for a MagicDotPath.
From psql docs: Tangent, argument in degrees.
See: https://www.postgresql.org/docs/current/functions-math.html
Parameters: |
|
---|
Returns: |
|
---|
Source code in py_linq_sql/utils/functions/magic_dp_trigo_functions.py
def tand(mdp: BaseMagicDotPath) -> MagicDotPathWithOp:
"""
Tand function for a MagicDotPath.
From psql docs: Tangent, argument in degrees.
See: https://www.postgresql.org/docs/current/functions-math.html
Args:
mdp: MagicDotPath on which we apply the tand.
Returns:
MagicDotPathWithOp with one operand and on the correct operator.
"""
return mdp._get_one_operand_operator( # pylint: disable=protected-access
TrigoFunctType.TAND,
)
MagicDotPath hyperbolic functions
All trigonometric functions for MagicDotPath.
acosh(mdp)
Acosh function for a MagicDotPath.
From psql docs: Inverse hyperbolic cosine.
See: https://www.postgresql.org/docs/current/functions-math.html
Parameters: |
|
---|
Returns: |
|
---|
Source code in py_linq_sql/utils/functions/magic_dp_hyperb_functions.py
def acosh(mdp: BaseMagicDotPath) -> MagicDotPathWithOp:
"""
Acosh function for a MagicDotPath.
From psql docs: Inverse hyperbolic cosine.
See: https://www.postgresql.org/docs/current/functions-math.html
Args:
mdp: MagicDotPath on which we apply the acosh.
Returns:
MagicDotPathWithOp with one operand and on the correct operator.
"""
return mdp._get_one_operand_operator( # pylint: disable=protected-access
HyperBFuncType.ACOSH,
)
asinh(mdp)
Asinh function for a MagicDotPath.
From psql docs: Inverse hyperbolic sine.
See: https://www.postgresql.org/docs/current/functions-math.html
Parameters: |
|
---|
Returns: |
|
---|
Source code in py_linq_sql/utils/functions/magic_dp_hyperb_functions.py
def asinh(mdp: BaseMagicDotPath) -> MagicDotPathWithOp:
"""
Asinh function for a MagicDotPath.
From psql docs: Inverse hyperbolic sine.
See: https://www.postgresql.org/docs/current/functions-math.html
Args:
mdp: MagicDotPath on which we apply the asinh.
Returns:
MagicDotPathWithOp with one operand and on the correct operator.
"""
return mdp._get_one_operand_operator( # pylint: disable=protected-access
HyperBFuncType.ASINH,
)
atanh(mdp)
Atanh function for a MagicDotPath.
From psql docs: Inverse hyperbolic tangent.
See: https://www.postgresql.org/docs/current/functions-math.html
Parameters: |
|
---|
Returns: |
|
---|
Source code in py_linq_sql/utils/functions/magic_dp_hyperb_functions.py
def atanh(mdp: BaseMagicDotPath) -> MagicDotPathWithOp:
"""
Atanh function for a MagicDotPath.
From psql docs: Inverse hyperbolic tangent.
See: https://www.postgresql.org/docs/current/functions-math.html
Args:
mdp: MagicDotPath on which we apply the atanh.
Returns:
MagicDotPathWithOp with one operand and on the correct operator.
"""
return mdp._get_one_operand_operator( # pylint: disable=protected-access
HyperBFuncType.ATANH,
)
cosh(mdp)
Cosh function for a MagicDotPath.
From psql docs: Hyperbolic cosine.
See: https://www.postgresql.org/docs/current/functions-math.html
Parameters: |
|
---|
Returns: |
|
---|
Source code in py_linq_sql/utils/functions/magic_dp_hyperb_functions.py
def cosh(mdp: BaseMagicDotPath) -> MagicDotPathWithOp:
"""
Cosh function for a MagicDotPath.
From psql docs: Hyperbolic cosine.
See: https://www.postgresql.org/docs/current/functions-math.html
Args:
mdp: MagicDotPath on which we apply the cosh.
Returns:
MagicDotPathWithOp with one operand and on the correct operator.
"""
return mdp._get_one_operand_operator( # pylint: disable=protected-access
HyperBFuncType.COSH,
)
sinh(mdp)
Sinh function for a MagicDotPath.
From psql docs: Hyperbolic sine.
See: https://www.postgresql.org/docs/current/functions-math.html
Parameters: |
|
---|
Returns: |
|
---|
Source code in py_linq_sql/utils/functions/magic_dp_hyperb_functions.py
def sinh(mdp: BaseMagicDotPath) -> MagicDotPathWithOp:
"""
Sinh function for a MagicDotPath.
From psql docs: Hyperbolic sine.
See: https://www.postgresql.org/docs/current/functions-math.html
Args:
mdp: MagicDotPath on which we apply the sinh.
Returns:
MagicDotPathWithOp with one operand and on the correct operator.
"""
return mdp._get_one_operand_operator( # pylint: disable=protected-access
HyperBFuncType.SINH,
)
tanh(mdp)
Tanh function for a MagicDotPath.
From psql docs: Hyperbolic tangent.
See: https://www.postgresql.org/docs/current/functions-math.html
Parameters: |
|
---|
Returns: |
|
---|
Source code in py_linq_sql/utils/functions/magic_dp_hyperb_functions.py
def tanh(mdp: BaseMagicDotPath) -> MagicDotPathWithOp:
"""
Tanh function for a MagicDotPath.
From psql docs: Hyperbolic tangent.
See: https://www.postgresql.org/docs/current/functions-math.html
Args:
mdp: MagicDotPath on which we apply the tanh.
Returns:
MagicDotPathWithOp with one operand and on the correct operator.
"""
return mdp._get_one_operand_operator( # pylint: disable=protected-access
HyperBFuncType.TANH,
)
Operators and functions for MagicDotPath
OperatorsType, functionsType aand the functions that use them.
col_name_hyper(name_op, operator)
Get the corresponding column name for an hyperbolic functions from a MagicDotPath.
Parameters: |
|
---|
Returns: |
|
---|
Exceptions: |
|
---|
Examples:
>>> col_name_hyper(DotMap(op1="mass"), HyperBFuncType.COSH)
'cosh_mass'
Source code in py_linq_sql/utils/classes/op_and_func_of_mdp.py
def col_name_hyper(name_op: DotMap, operator: _OPERATORTYPE) -> str:
"""
Get the corresponding column name for an hyperbolic functions from a MagicDotPath.
Args:
name_op: name of the operand(s).
operator: Operator Type.
Returns:
A column name with the correct format.
Raises:
TypeOperatorError: If type of operator is not `HyperBFuncType`.
Examples:
>>> col_name_hyper(DotMap(op1="mass"), HyperBFuncType.COSH)
'cosh_mass'
"""
if not isinstance(operator, HyperBFuncType):
raise TypeOperatorError([HyperBFuncType], type(operator))
return f"{operator.name}_{name_op.op1}"
col_name_math(name_op, operator)
Get the corresponding column name for a mathematical functions from a MagicDotPath.
Parameters: |
|
---|
Returns: |
|
---|
Exceptions: |
|
---|
Examples:
>>> col_name_math(DotMap(op1="mass"), MathFunctType.CEIL)
'ceil_mass'
Source code in py_linq_sql/utils/classes/op_and_func_of_mdp.py
def col_name_math(name_op: DotMap, operator: _OPERATORTYPE) -> str:
"""
Get the corresponding column name for a mathematical functions from a MagicDotPath.
Args:
name_op: name of the operand(s).
operator: Operator Type.
Returns:
A column name with the correct format.
Raises:
TypeOperatorError: If type of operator is not `MathFunctType`.
Examples:
>>> col_name_math(DotMap(op1="mass"), MathFunctType.CEIL)
'ceil_mass'
"""
if not isinstance(operator, MathFunctType):
raise TypeOperatorError([MathFunctType], type(operator))
if operator in [
MathFunctType.GCD,
MathFunctType.LCM,
MathFunctType.LOG,
MathFunctType.ROUND,
MathFunctType.TRUNC,
MathFunctType.GREATEST,
MathFunctType.LEAST,
]:
return f"{operator.name}_{name_op.op1}_{name_op.op2}"
return f"{operator.name}_{name_op.op1}"
col_name_ope(name_op, operator)
Get the corresponding column name for an operator from a MagicDotPath.
Parameters: |
|
---|
Returns: |
|
---|
Exceptions: |
|
---|
Examples:
>>> col_name_ope(DotMap(op1="mass", op2="power"), OperatorType.POW)
'mass_pow_power'
Source code in py_linq_sql/utils/classes/op_and_func_of_mdp.py
def col_name_ope(name_op: DotMap, operator: _OPERATORTYPE) -> str:
"""
Get the corresponding column name for an operator from a MagicDotPath.
Args:
name_op: name of the operand(s).
operator: Operator Type.
Returns:
A column name with the correct format.
Raises:
TypeOperatorError: If type of operator is not `OperatorType.`.
Examples:
>>> col_name_ope(DotMap(op1="mass", op2="power"), OperatorType.POW)
'mass_pow_power'
"""
if not isinstance(operator, OperatorType):
raise TypeOperatorError([OperatorType], type(operator))
if operator in [
OperatorType.ABS,
OperatorType.NEG,
OperatorType.POS,
OperatorType.INVERT,
]:
return f"{operator.name}_{name_op.op1}"
return f"{name_op.op1}_{operator.name}_{name_op.op2}"
col_name_str(name_op, operator)
Get the corresponding column name for a str functions from a MagicDotPath.
Parameters: |
|
---|
Returns: |
|
---|
Exceptions: |
|
---|
Examples:
>>> col_name_str(DotMap(op1="star_name"), StrFunctType.LOWER)
>>> 'lower_star_name'
Source code in py_linq_sql/utils/classes/op_and_func_of_mdp.py
def col_name_str(name_op: DotMap, operator: _OPERATORTYPE) -> str:
"""
Get the corresponding column name for a str functions from a MagicDotPath.
Args:
name_op: name of the operand(s).
operator: Operator Type.
Returns:
A column name with correct format.
Raises:
TypeOperatorErrir: If type of operator is not `StrFunctType`.
Examples:
>>> col_name_str(DotMap(op1="star_name"), StrFunctType.LOWER)
>>> 'lower_star_name'
"""
if not isinstance(operator, StrFunctType):
raise TypeOperatorError([StrFunctType], type(operator))
return f"{operator.name}_{name_op.op1}"
col_name_trigo(name_op, operator)
Get the corresponding column name for a trigonometric functions from a MagicDotPath.
Parameters: |
|
---|
Returns: |
|
---|
Exceptions: |
|
---|
Examples:
>>> col_name_trigo(DotMap(op1="mass"), TrigoFunctType.ASIND)
>>> "asind_mass"
Source code in py_linq_sql/utils/classes/op_and_func_of_mdp.py
def col_name_trigo(name_op: DotMap, operator: _OPERATORTYPE) -> str:
"""
Get the corresponding column name for a trigonometric functions from a MagicDotPath.
Args:
name_op: name of the operand(s).
operator: Operator Type.
Returns:
A column name with the correct format.
Raises:
TypeOperatorError: If type of operator is not `TrigoFunctType`.
Examples:
>>> col_name_trigo(DotMap(op1="mass"), TrigoFunctType.ASIND)
>>> "asind_mass"
"""
if not isinstance(operator, TrigoFunctType):
raise TypeOperatorError([TrigoFunctType], type(operator))
if operator in [TrigoFunctType.ATAN2, TrigoFunctType.ATAN2D]:
return f"{operator.name}_{name_op.op1}_{name_op.op2}"
return f"{operator.name}_{name_op.op1}"
json_path_hyper(path_op, operator)
Get the corresponding jsonb path for an hyperbolic function from a MagicDotPath.
Parameters: |
|
---|
Returns: |
|
---|
Exceptions: |
|
---|
Source code in py_linq_sql/utils/classes/op_and_func_of_mdp.py
def json_path_hyper(path_op: DotMap, operator: _OPERATORTYPE) -> str:
"""
Get the corresponding jsonb path for an hyperbolic function from a MagicDotPath.
Args:
path_op: path of the operand(s).
operator: Operator Type.
Returns:
A path with the correct jsonb syntax.
Raises:
TypeOperatorError: If type of operator is not `HyperBFuncType`.
"""
if not isinstance(operator, HyperBFuncType):
raise TypeOperatorError([HyperBFuncType], type(operator))
return f"{operator.psql}(CAST({path_op.op1} AS decimal))"
json_path_math(path_op, operator)
Get the corresponding jsonb path for a mathematical function from a MagicDotPath.
Parameters: |
|
---|
Returns: |
|
---|
Exceptions: |
|
---|
Source code in py_linq_sql/utils/classes/op_and_func_of_mdp.py
def json_path_math(path_op: DotMap, operator: _OPERATORTYPE) -> str:
"""
Get the corresponding jsonb path for a mathematical function from a MagicDotPath.
Args:
path_op: path of the operand(s).
operator: Operator Type.
Returns:
A path with the correct jsonb syntax.
Raises:
TypeOperatorError: If type of operator is not `MathFunctType`.
"""
if not isinstance(operator, MathFunctType):
raise TypeOperatorError([MathFunctType], type(operator))
if operator in [MathFunctType.FACTORIAL]:
return f"{operator.psql}(CAST({path_op.op1} AS integer))"
if operator in [
MathFunctType.SQRT,
MathFunctType.CBRT,
]:
return f"{operator.psql} CAST({path_op.op1} AS decimal)"
if operator in [MathFunctType.GCD, MathFunctType.LCM]:
operand_1 = f"CAST({path_op.op1} AS integer)"
operand_2 = f"CAST({path_op.op2} AS integer)"
return f"{operator.psql}({operand_1}, {operand_2})"
if operator in [MathFunctType.TRUNC, MathFunctType.ROUND]:
return f"{operator.psql}(CAST({path_op.op1} AS decimal), {path_op.op2})"
if operator in [MathFunctType.LOG]:
return f"{operator.psql}({path_op.op2}, CAST({path_op.op1} AS decimal))"
if operator in [MathFunctType.GREATEST, MathFunctType.LEAST]:
operand_1 = f"CAST({path_op.op1} AS decimal)"
operand_2 = f"CAST({path_op.op2} AS decimal)"
return f"{operator.psql}({operand_1}, {operand_2})"
return f"{operator.psql}(CAST({path_op.op1} AS decimal))"
json_path_ope(path_op, operator)
Get the corresponding jsonb path for an operator from a MagicDotPath.
Parameters: |
|
---|
Returns: |
|
---|
Exceptions: |
|
---|
Source code in py_linq_sql/utils/classes/op_and_func_of_mdp.py
def json_path_ope(path_op: DotMap, operator: _OPERATORTYPE) -> str:
"""
Get the corresponding jsonb path for an operator from a MagicDotPath.
Args:
path_op: path of the operand(s).
operator: Operator Type.
Returns:
A path with the correct jsonb syntax.
Raises:
TypeOperatorError: If type of operator is not `OperatorType`.
"""
if not isinstance(operator, OperatorType):
raise TypeOperatorError([OperatorType], type(operator))
if operator in [OperatorType.AND, OperatorType.OR, OperatorType.IN]:
return f"{path_op.op1} {operator.psql} {path_op.op2}"
if operator in [OperatorType.INVERT]:
return f"{operator.psql} {path_op.op1}"
return f"{operator.psql} CAST({path_op.op1} AS decimal)"
json_path_str(path_op, operator)
Get the corresponding jsonb path for a str function from a MagicDotPath.
Parameters: |
|
---|
Returns: |
|
---|
Exceptions: |
|
---|
Source code in py_linq_sql/utils/classes/op_and_func_of_mdp.py
def json_path_str(path_op: DotMap, operator: _OPERATORTYPE) -> str:
"""
Get the corresponding jsonb path for a str function from a MagicDotPath.
Args:
path_op: path of the operand(s).
operator: Operator Type.
Returns:
A path with the correct jsonb syntax.
Raises:
TypeOperatorError: If type of operator is not `StrFunctType`.
"""
if not isinstance(operator, StrFunctType):
raise TypeOperatorError([StrFunctType], type(operator))
return f"{operator.psql}(CAST({path_op.op1} AS TEXT))"
json_path_trigo(path_op, operator)
Get the corresponding jsonb path for a trigonometric function from a MagicDotPath.
Parameters: |
|
---|
Returns: |
|
---|
Exceptions: |
|
---|
Source code in py_linq_sql/utils/classes/op_and_func_of_mdp.py
def json_path_trigo(path_op: DotMap, operator: _OPERATORTYPE) -> str:
"""
Get the corresponding jsonb path for a trigonometric function from a MagicDotPath.
Args:
path_op: path of the operand(s).
operator: Operator Type.
Returns:
A path with the correct jsonb syntax.
Raises:
TypeOperatorError: If type of operator is not `TrigoFunctType`.
"""
if not isinstance(operator, TrigoFunctType):
raise TypeOperatorError([TrigoFunctType], type(operator))
if operator in [TrigoFunctType.ATAN2, TrigoFunctType.ATAN2D]:
return f"{operator.psql}(CAST({path_op.op1} AS decimal), {path_op.op2})"
return f"{operator.psql}(CAST({path_op.op1} AS decimal))"
Other functions
Functions used in py-linq-sql.
get_columns_name(mdps)
Get all column name.
Parameters: |
|
---|
Returns: |
|
---|
Exceptions: |
|
---|
Source code in py_linq_sql/utils/functions/other_functions.py
def get_columns_name(
mdps: MagicDotPath
| List[MagicDotPath]
| Tuple[MagicDotPath]
| Dict[str, MagicDotPath],
) -> List[str]:
"""
Get all column name.
Args:
mdps: MagicDotPath for which we want the column name.
Returns:
All of paths and columns.
Raises:
TypeOperatorError: Indirect raise by `MagicDotPathWithOp.col_name`.
"""
result = []
match mdps:
case BaseMagicDotPath():
col_name = mdps.col_name()
result.append(
f"_{_short_columns_default_name(col_name)}"
if len(col_name) > 58
else f"{col_name}",
)
case tuple() | list():
for element in mdps:
col_name = element.col_name()
result.append(
f"_{_short_columns_default_name(col_name)}"
if len(col_name) > 58
else f"{col_name}",
)
case dict():
for key in mdps:
result.append(_col_name_validator(key))
return _fix_same_column_name(result)
get_good_type(cast_type)
Get the good type as str from a cast_type.
Parameters: |
|
---|
Returns: |
|
---|
Source code in py_linq_sql/utils/functions/other_functions.py
def get_good_type(cast_type: type) -> str | None:
"""
Get the good type as str from a cast_type.
Args:
cast_type: Type in which we want to cast the path.
Returns:
SQL type with '))' to cast in sql command. None if we give the wrong type.
"""
if cast_type in (int, float, Decimal):
return "decimal))"
if cast_type == date:
return "date))"
return None
get_json(data)
Get a json from data.
Parameters: |
|
---|
Returns: |
|
---|
Exceptions: |
|
---|
Source code in py_linq_sql/utils/functions/other_functions.py
def get_json(data: Dict[Any, Any]) -> str:
"""
Get a json from data.
Args:
data: data we want to have in json.
Returns:
A json contains data.
Raises:
ValueError: Indirect raise by `json.dumps`.
"""
return json.dumps(data)
pretty_print(record)
Print a record in a pretty table with rich.
Parameters: |
|
---|
Exceptions: |
|
---|
Source code in py_linq_sql/utils/functions/other_functions.py
def pretty_print(record: Enumerable) -> None:
"""
Print a record in a pretty table with rich.
Args:
record: Record to display.
Raises:
rich.errors.ConsoleError: Indirect raise by `rich.table`
rich.console.CaptureError: Indirect raise by `rich.console`
"""
if record is None:
Console().print("Record is None.")
else:
rec_in_list = record.to_list()
if not rec_in_list:
Console().print("Empty Enumerable.")
else:
table = Table(title="Record", show_lines=True)
for name in rec_in_list[0]._fields:
table.add_column(name, justify="center", no_wrap=True)
for element in rec_in_list:
table.add_row(*map(str, element))
Console().print(table)
count = len(rec_in_list)
row_repr = f"({count} rows)" if count > 1 else f"({count} row)"
Console().print(row_repr)
safe(connection, name)
Secure a column or a table for a request.
Parameters: |
|
---|
Returns: |
|
---|
Exceptions: |
|
---|
Source code in py_linq_sql/utils/functions/other_functions.py
def safe(connection: Connection, name: str) -> str:
"""
Secure a column or a table for a request.
Args:
name: Name of the column or table we want to secure.
Returns:
Name but verified by psycopg.sql.Identifier.
Raises:
psycopg.Error: Indirect raise by `sql.Identifier` or `as_string`.
"""
return sql.Identifier(name).as_string(connection)
Classes API
Command class
Class of command with command type and a string with a command of this type.
Attributes:
Name | Type | Description |
---|---|---|
cmd_type |
CommandType |
Type of command. |
args |
DotMap |
All arguments for this command. |
Source code in py_linq_sql/utils/classes/other_classes.py
@dataclass
class Command:
"""
Class of command with command type and a string with a command of this type.
Attributes:
cmd_type (CommandType): Type of command.
args (DotMap): All arguments for this command.
"""
cmd_type: CommandTypeOrStr
args: DotMap = field(default_factory=DotMap)
def __eq__(self, other: Any) -> bool: # noqa: ANN401
"""
Try equality between two Command.
Command_1 == Command_2.
"""
if not isinstance(other, Command):
return NotImplemented
if not self.cmd_type == other.cmd_type:
return False
for key in self.args:
if not self.args[key] == other.args[key]:
return False
return True
def __ne__(self, other: Any) -> bool: # noqa: ANN401
"""
Try no-equality between two Command.
Command_1 != Command_2.
"""
return bool(not self.__eq__(other))
__eq__(self, other)
special
Try equality between two Command.
Command_1 == Command_2.
Source code in py_linq_sql/utils/classes/other_classes.py
def __eq__(self, other: Any) -> bool: # noqa: ANN401
"""
Try equality between two Command.
Command_1 == Command_2.
"""
if not isinstance(other, Command):
return NotImplemented
if not self.cmd_type == other.cmd_type:
return False
for key in self.args:
if not self.args[key] == other.args[key]:
return False
return True
__ne__(self, other)
special
Try no-equality between two Command.
Command_1 != Command_2.
Source code in py_linq_sql/utils/classes/other_classes.py
def __ne__(self, other: Any) -> bool: # noqa: ANN401
"""
Try no-equality between two Command.
Command_1 != Command_2.
"""
return bool(not self.__eq__(other))
Enum
Enum used in py-linq-sql.
CommandType (StrEnum)
Enum of command type.
Source code in py_linq_sql/utils/classes/enum.py
class CommandType(StrEnum):
"""Enum of command type."""
SELECT = "select"
WHERE = "where"
MIN = "min"
MAX = "max"
INSERT = "insert"
ORDER_BY = "order by"
ORDER_BY_DESC = "order by desc"
TAKE = "take"
SKIP = "skip"
ELEMENT_AT = "element at"
FIRST = "first"
UPDATE = "update"
COUNT = "count"
SINGLE = "single"
LAST = "last"
TAKE_LAST = "take last"
SKIP_LAST = "skip last"
ALL = "all"
ANY = "any"
CONTAINS = "contains"
EXCEPT_ = "except"
JOIN = "join"
UNION = "union"
INTERSECT = "intersect"
DISTINCT = "distinct"
GROUP_BY = "group by"
DELETE = "delete"
GROUP_JOIN = "group join"
JoinType (_Type, Enum)
Enum of join type.
Source code in py_linq_sql/utils/classes/enum.py
class JoinType(_Type, Enum):
"""Enum of join type."""
INNER = _Type(is_intersect=False, as_str="INNER")
LEFT = _Type(False, "LEFT")
LEFT_MINUS_INTERSECT = _Type(True, "LEFT")
RIGHT = _Type(False, "RIGHT")
RIGHT_MINUS_INTERSECT = _Type(True, "RIGHT")
FULL = _Type(False, "FULL")
FULL_MINUS_INTERSECT = _Type(True, "FULL")
__new__(cls, value)
special
Create and return a new object. See help(type) for accurate signature.
Source code in py_linq_sql/utils/classes/enum.py
def __new__(cls, value):
# all enum instances are actually created during class construction
# without calling this method; this method is called by the metaclass'
# __call__ (i.e. Color(3) ), and by pickle
if type(value) is cls:
# For lookups like Color(Color.RED)
return value
# by-value search for a matching enum member
# see if it's in the reverse mapping (for hashable values)
try:
return cls._value2member_map_[value]
except KeyError:
# Not found, no need to do long O(n) search
pass
except TypeError:
# not there, now do long search -- O(n) behavior
for member in cls._member_map_.values():
if member._value_ == value:
return member
# still not found -- verify that members exist, in-case somebody got here mistakenly
# (such as via super when trying to override __new__)
if not cls._member_map_:
raise TypeError("%r has no members defined" % cls)
#
# still not found -- try _missing_ hook
try:
exc = None
result = cls._missing_(value)
except Exception as e:
exc = e
result = None
try:
if isinstance(result, cls):
return result
elif (
Flag is not None and issubclass(cls, Flag)
and cls._boundary_ is EJECT and isinstance(result, int)
):
return result
else:
ve_exc = ValueError("%r is not a valid %s" % (value, cls.__qualname__))
if result is None and exc is None:
raise ve_exc
elif exc is None:
exc = TypeError(
'error in %s._missing_: returned %r instead of None or a valid member'
% (cls.__name__, result)
)
if not isinstance(exc, ValueError):
exc.__context__ = ve_exc
raise exc
finally:
# ensure all variables that could hold an exception are destroyed
exc = None
ve_exc = None
__repr__(self)
special
Return repr(self).
Source code in py_linq_sql/utils/classes/enum.py
def __repr__(self):
v_repr = self.__class__._value_repr_ or repr
return "<%s.%s: %s>" % (self.__class__.__name__, self._name_, v_repr(self._value_))
Terminal (Enum)
Enum of state of terminal flags.
Source code in py_linq_sql/utils/classes/enum.py
class Terminal(Enum):
"""Enum of state of terminal flags."""
COUNT = auto()
DISTINCT = auto()
ELEMENT_AT = auto()
EXCEPT_ = auto()
INTERSECT = auto()
LAST = auto()
MAX = auto()
MIN = auto()
SINGLE = auto()
UNION = auto()
GROUP_BY = auto()
GROUP_JOIN = auto()
Flags class
Class of flags for SQLEnumerable.
Attributes:
Name | Type | Description |
---|---|---|
select |
bool |
True if we call |
alter |
bool |
True if we call an alter methods (insert, update), False otherwise. |
one |
bool |
True if we call a one methods (all, any, contains), False otherwise. |
terminal |
Terminal |
Type of terminal command. |
limit_offset |
bool |
True if we call a limit offset methods (see docs), False otherwise. |
join |
bool |
True if we call join methods (see docs), False otherwise. |
default_cmd |
bool |
True if we call default methods (see docs), False otherwise. |
Source code in py_linq_sql/utils/classes/other_classes.py
@dataclass
class Flags:
"""
Class of flags for SQLEnumerable.
Attributes:
select (bool): True if we call `select()` method, False otherwise.
alter (bool): True if we call an alter methods (insert, update),
False otherwise.
one (bool): True if we call a one methods (all, any, contains), False otherwise.
terminal (Terminal): Type of terminal command.
limit_offset (bool): True if we call a limit offset methods (see docs),
False otherwise.
join (bool): True if we call join methods (see docs), False otherwise.
default_cmd (bool): True if we call default methods (see docs),
False otherwise.
"""
select: bool = False
alter: bool = False
one: bool = False
terminal: Terminal | None = None
limit_offset: bool = False
join: bool = False
default_cmd: bool = False
def __eq__(self, other: Any) -> bool: # noqa: ANN401
"""
Try equality between two Flags.
Flags_1 == Flags_2.
"""
return equality(self, other)
def __ne__(self, other: Any) -> bool: # noqa: ANN401
"""
Try no-equality between two Flags.
Flags_1 != Flags_2.
"""
return bool(not self.__eq__(other))
def copy(self) -> Flags:
"""Create a shallow copy of self."""
return Flags(
self.select,
self.alter,
self.one,
self.terminal,
self.limit_offset,
self.join,
self.default_cmd,
)
__eq__(self, other)
special
Try equality between two Flags.
Flags_1 == Flags_2.
Source code in py_linq_sql/utils/classes/other_classes.py
def __eq__(self, other: Any) -> bool: # noqa: ANN401
"""
Try equality between two Flags.
Flags_1 == Flags_2.
"""
return equality(self, other)
__ne__(self, other)
special
Try no-equality between two Flags.
Flags_1 != Flags_2.
Source code in py_linq_sql/utils/classes/other_classes.py
def __ne__(self, other: Any) -> bool: # noqa: ANN401
"""
Try no-equality between two Flags.
Flags_1 != Flags_2.
"""
return bool(not self.__eq__(other))
copy(self)
Create a shallow copy of self.
Source code in py_linq_sql/utils/classes/other_classes.py
def copy(self) -> Flags:
"""Create a shallow copy of self."""
return Flags(
self.select,
self.alter,
self.one,
self.terminal,
self.limit_offset,
self.join,
self.default_cmd,
)
MagicDotPath classes
BaseMagicDotPath
Abstract base for Magical object with operator methods.
Inspired by The very useful DotMap module this
object allow to write predicate in the lambda function used in SQL clause like
.select()
or .join()
. This is very useful to express SQL constraint.
See the LINQ documentation for more in depth explanation.
Source code in py_linq_sql/utils/classes/magicdotpath.py
class BaseMagicDotPath:
"""
Abstract base for Magical object with operator methods.
Inspired by The very useful [DotMap module](https://github.com/drgrib/dotmap) this
object allow to write predicate in the lambda function used in SQL clause like
`.select()` or `.join()`. This is very useful to express SQL constraint.
See the LINQ documentation for more in depth explanation.
"""
def __getattr__(self, attribute_name: str) -> BaseMagicDotPath:
"""
Get attribute in a list for a MagicDotPath objects.
Args:
attribute_name: Names of all attributes.
Returns:
A MagicDotPath objects with attributes names in attributes list.
Raises:
NotImplementedError: This function is just a wrapper for the subclasses.
"""
# No cover because is just a wrapper for subclasses.
raise NotImplementedError # pragma: no cover
def __getitem__(self, attribute_name: str) -> BaseMagicDotPath:
"""
Get items in a list for a MagicDotPath objects.
Args:
attribute_name: Names of all attributes.
Returns:
A MagicDotPath objects with attributes names in attributes list.
Raises:
NotImplementedError: This function is just a wrapper for the subclasses.
"""
# No cover because is just a wrapper for subclasses.
raise NotImplementedError # pragma: no cover
def _get_number_operator(
self,
other: _VALID_OPERAND_TYPE | BaseMagicDotPath,
operator: _OPERATORTYPE,
) -> MagicDotPathWithOp:
"""
Get a MagicDotPathWithOp objects with a number operator.
Args:
other: Second objects for the comparison.
operator: type of the operator.
Returns:
A MagicDotPathWithOp contains two operand and an operator
with the correct type.
Raises:
TypeOperatorError: If the type of other is not accepted by the function.
Accepted type: int, float, decimal, BaseMagicDotPath.
"""
other_type: Type[Any]
match other:
case float():
other_type = float
case int():
other_type = int
case Decimal():
other_type = Decimal
case BaseMagicDotPath():
other_type = BaseMagicDotPath
case _:
raise TypeOperatorError(
[int, float, Decimal, BaseMagicDotPath],
type(other),
)
return MagicDotPathWithOp(
operand_1=self,
operator=operator,
operand_2=other,
my_type=other_type,
)
def _get_generic_operator(
self,
other: _VALID_OPERAND_TYPE | BaseMagicDotPath,
operator: _OPERATORTYPE,
) -> MagicDotPathWithOp:
"""
Get a MagicDotPathWithOp objects with a generic operator.
Args:
other: Second objects for the comparison.
operator: type of the operator.
Returns:
A MagicDotPathWithOp contains two operand and an operator
with the correct type.
Raises:
TypeOperatorError: If the type of other is not accepted by the function.
Accepted type: int, float, decimal, str, list, dict, BaseMagicDotPath.
"""
other_type: Type[Any] | None = None
match other:
case float():
other_type = float
case bool():
other_type = bool
case int():
other_type = int
case Decimal():
other_type = Decimal
case str():
other_type = str
case list():
if operator == OperatorType.IN:
other_type = str
else:
other_type = list
case dict():
other_type = dict
case BaseMagicDotPath():
other_type = BaseMagicDotPath
case _:
raise TypeOperatorError(
[float, int, Decimal, str, list, dict, bool, BaseMagicDotPath],
type(other),
)
return MagicDotPathWithOp(
operand_1=self,
operator=operator,
operand_2=other,
my_type=other_type,
)
def _get_one_operand_operator(self, operator: _OPERATORTYPE) -> MagicDotPathWithOp:
"""
Get the operator for single operand MagicDotPathWithOp (~, abs, ...).
Args:
operator: The operator we want.
Returns:
A MagicDotPathWithOp contains an operand and an operator, operand_2 is None,
with `~(x.data.obj.name) #not x.data.obj.name` we have only one operand and
`my_type` is NoneType therefore.
"""
return MagicDotPathWithOp(
operand_1=self,
operator=operator,
operand_2=None,
my_type=NoneType,
)
def __gt__(
self,
other: _VALID_OPERAND_TYPE | BaseMagicDotPath,
) -> MagicDotPathWithOp:
"""
Get a MagicDotPathWithOp objects with a `>` operator and the correct types.
Types are integer, float, Decimal or BaseMagicDotPath.
Args:
other: The value at which the comparison is made.
Returns:
MagicDotPathWithOp.
Raises:
TypeOperatorError: Indirect raise by `_get_number_operator`
"""
return self._get_number_operator(other, OperatorType.GT)
def __lt__(
self,
other: _VALID_OPERAND_TYPE | BaseMagicDotPath,
) -> MagicDotPathWithOp:
"""
Get a MagicDotPathWithOp objects with a `<` operator and the correct types.
Types are integer, float, Decimal or BaseMagicDotPath.
Args:
other: The value at which the comparison is made.
Returns:
MagicDotPathWithOp.
Raises:
TypeOperatorError: Indirect raise by `_get_number_operator`
"""
return self._get_number_operator(other, OperatorType.LT)
def __ge__(
self,
other: _VALID_OPERAND_TYPE | BaseMagicDotPath,
) -> MagicDotPathWithOp:
"""
Get a MagicDotPathWithOp objects with a `>=` operator and the correct types.
Types are integer, float, Decimal or BaseMagicDotPath.
Args:
other: The value at which the comparison is made.
Returns:
MagicDotPathWithOp.
Raises:
TypeOperatorError: Indirect raise by `_get_number_operator`
"""
return self._get_number_operator(other, OperatorType.GE)
def __le__(
self,
other: _VALID_OPERAND_TYPE | BaseMagicDotPath,
) -> MagicDotPathWithOp:
"""
Get a MagicDotPathWithOp objects with a `<=` operator and the correct types.
Types are integer, float, Decimal or BaseMagicDotPath.
Args:
other: The value at which the comparison is made.
Returns:
MagicDotPathWithOp.
Raises:
TypeOperatorError: Indirect raise by `_get_number_operator`
"""
return self._get_number_operator(other, OperatorType.LE)
def __eq__( # type: ignore[override]
self,
other: _VALID_OPERAND_TYPE | BaseMagicDotPath,
) -> MagicDotPathWithOp:
"""
Get a MagicDotPathWithOp objects with a `=` operator and the correct types.
Types are integer, float, Decimal, string, list, dict, BaseMagicDotPath.
Args:
other: The value at which the comparison is made.
Returns:
MagicDotPathWithOp.
Raises:
TypeOperatorError: Indirect raise by `_get_generic_operator`
"""
return self._get_generic_operator(other, OperatorType.EQ)
def __ne__( # type: ignore[override]
self,
other: _VALID_OPERAND_TYPE | BaseMagicDotPath,
) -> MagicDotPathWithOp:
"""
Get a MagicDotPathWithOp objects with a `<>` operator and the correct types.
Types are integer, float, Decimal, string, list, dict, BaseMagicDotPath.
Args:
other: The value at which the comparison is made.
Returns:
MagicDotPathWithOp.
Raises:
TypeOperatorError: Indirect raise by `_get_generic_operator`
"""
return self._get_generic_operator(other, OperatorType.NE)
def __add__(
self,
other: _VALID_OPERAND_TYPE | BaseMagicDotPath,
) -> MagicDotPathWithOp:
"""
Get a MagicDotPathWithOp objects with a `+` operator and the correct types.
Types are integer, float, Decimal, BaseMagicDotPath.
Args:
other: The value at which the comparison is made.
Returns:
MagicDotPathWithOp.
Raises:
TypeOperatorError: Indirect raise by `_get_number_operator`
"""
return self._get_number_operator(other, OperatorType.ADD)
def __radd__(
self,
other: _VALID_OPERAND_TYPE | BaseMagicDotPath,
) -> MagicDotPathWithOp:
"""
Get a MagicDotPathWithOp objects with a `+` operator and the correct types.
This operator is useful when we make `8 + MagicDotPath`
and not `MagicDotPath + 8`.
Types are integer, float, Decimal, BaseMagicDotPath.
Args:
other: The value at which the comparison is made.
Returns:
MagicDotPathWithOp.
Raises:
TypeOperatorError: Indirect raise by `_get_number_operator`
"""
return self.__add__(other)
def __sub__(
self,
other: _VALID_OPERAND_TYPE | BaseMagicDotPath,
) -> MagicDotPathWithOp:
"""
Get a MagicDotPathWithOp objects with a `-` operator and the correct types.
Types are integer, float, Decimal, BaseMagicDotPath.
Args:
other: The value at which the comparison is made.
Returns:
MagicDotPathWithOp.
Raises:
TypeOperatorError: Indirect raise by `_get_number_operator`
"""
return self._get_number_operator(other, OperatorType.SUB)
def __rsub__(
self,
other: _VALID_OPERAND_TYPE | BaseMagicDotPath,
) -> MagicDotPathWithOp:
"""
Get a MagicDotPathWithOp objects with a `-` operator and the correct types.
This operator is useful when we make `8 - MagicDotPath`
and not `MagicDotPath - 8`.
Types are integer, float, Decimal, BaseMagicDotPath.
Args:
other: The value at which the comparison is made.
Returns:
MagicDotPathWithOp.
Raises:
TypeOperatorError: Indirect raise by `_get_number_operator`
"""
mdp_with_op = self._get_number_operator(self, OperatorType.SUB)
mdp_with_op.operand_1 = other
return mdp_with_op
def __mul__(
self,
other: _VALID_OPERAND_TYPE | BaseMagicDotPath,
) -> MagicDotPathWithOp:
"""
Get a MagicDotPathWithOp objects with a `*` operator and the correct types.
Types are integer, float, Decimal, BaseMagicDotPath.
Args:
other: The value at which the comparison is made.
Returns:
MagicDotPathWithOp.
Raises:
TypeOperatorError: Indirect raise by `_get_number_operator`
"""
return self._get_number_operator(other, OperatorType.MUL)
def __rmul__(
self,
other: _VALID_OPERAND_TYPE | BaseMagicDotPath,
) -> MagicDotPathWithOp:
"""
Get a MagicDotPathWithOp objects with a `*` operator and the correct types.
This operator is useful when we make `8 * MagicDotPath`
and not `MagicDotPath * 8`.
Types are integer, float, Decimal, BaseMagicDotPath.
Args:
other: The value at which the comparison is made.
Returns:
MagicDotPathWithOp.
Raises:
TypeOperatorError: Indirect raise by `_get_number_operator`
"""
return self.__mul__(other)
def __truediv__(
self,
other: _VALID_OPERAND_TYPE | BaseMagicDotPath,
) -> MagicDotPathWithOp:
"""
Get a MagicDotPathWithOp objects with a `/` operator and the correct types.
Types are integer, float, Decimal, BaseMagicDotPath.
Args:
other: The value at which the comparison is made.
Returns:
MagicDotPathWithOp.
Raises:
TypeOperatorError: Indirect raise by `_get_number_operator`
"""
return self._get_number_operator(other, OperatorType.TRUEDIV)
def __rtruediv__(
self,
other: _VALID_OPERAND_TYPE | BaseMagicDotPath,
) -> MagicDotPathWithOp:
"""
Get a MagicDotPathWithOp objects with a `/` operator and the correct types.
This operator is useful when we make `8 / MagicDotPath`
and not `MagicDotPath / 8`.
Types are integer, float, Decimal, BaseMagicDotPath.
Args:
other: The value at which the comparison is made.
Returns:
MagicDotPathWithOp.
Raises:
TypeOperatorError: Indirect raise by `_get_number_operator`
"""
mdp_with_op = self._get_number_operator(self, OperatorType.TRUEDIV)
mdp_with_op.operand_1 = other
return mdp_with_op
def __mod__(
self,
other: _VALID_OPERAND_TYPE | BaseMagicDotPath,
) -> MagicDotPathWithOp:
"""
Get a MagicDotPathWithOp objects with a `%` operator and the correct types.
Types are integer, float, Decimal, BaseMagicDotPath.
Args:
other: The value at which the comparison is made.
Returns:
MagicDotPathWithOp.
Raises:
TypeOperatorError: Indirect raise by `_get_number_operator`
"""
return self._get_number_operator(other, OperatorType.MOD)
def __rmod__(
self,
other: _VALID_OPERAND_TYPE | BaseMagicDotPath,
) -> MagicDotPathWithOp:
"""
Get a MagicDotPathWithOp objects with a `%` operator and the correct types.
This operator is useful when we make `8 % MagicDotPath`
and not `MagicDotPath % 8`.
Types are integer, float, Decimal, BaseMagicDotPath.
Args:
other: The value at which the comparison is made.
Returns:
MagicDotPathWithOp.
Raises:
TypeOperatorError: Indirect raise by `_get_number_operator`
"""
mdp_with_op = self._get_number_operator(self, OperatorType.MOD)
mdp_with_op.operand_1 = other
return mdp_with_op
def __and__(
self,
other: _VALID_OPERAND_TYPE | BaseMagicDotPath,
) -> MagicDotPathWithOp:
"""
Get a MagicDotPathWithOp objects with a `AND` operator and the correct types.
Types are integer, float, Decimal, BaseMagicDotPath.
Args:
other: The value at which the comparison is made.
Returns:
MagicDotPathWithOp.
Raises:
TypeOperatorError: Indirect raise by `_get_generic_operator`
"""
return self._get_generic_operator(other, OperatorType.AND)
def __rand__(
self,
other: _VALID_OPERAND_TYPE | BaseMagicDotPath,
) -> MagicDotPathWithOp:
"""
Get a MagicDotPathWithOp objects with a `AND` operator and the correct types.
This operator is useful when we make `8 & MagicDotPath`
and not `MagicDotPath & 8`.
Types are integer, float, Decimal, BaseMagicDotPath.
Args:
other: The value at which the comparison is made.
Returns:
MagicDotPathWithOp.
Raises:
TypeOperatorError: Indirect raise by `_get_generic_operator`
"""
return self.__and__(other)
def __or__(
self,
other: _VALID_OPERAND_TYPE | BaseMagicDotPath,
) -> MagicDotPathWithOp:
"""
Get a MagicDotPathWithOp objects with a `OR` operator and the correct types.
Types are integer, float, Decimal, BaseMagicDotPath.
Args:
other: The value at which the comparison is made.
Returns:
MagicDotPathWithOp.
Raises:
TypeOperatorError: Indirect raise by `_get_generic_operator`
"""
return self._get_generic_operator(other, OperatorType.OR)
def __ror__(
self,
other: _VALID_OPERAND_TYPE | BaseMagicDotPath,
) -> MagicDotPathWithOp:
"""
Get a MagicDotPathWithOp objects with a `OR` operator and the correct types.
This operator is useful when we make `8 | MagicDotPath`
and not `MagicDotPath | 8`.
Types are integer, float, Decimal, BaseMagicDotPath.
Args:
other: The value at which the comparison is made.
Returns:
MagicDotPathWithOp.
Raises:
TypeOperatorError: Indirect raise by `_get_generic_operator`
"""
return self.__or__(other)
def __invert__(self) -> MagicDotPathWithOp:
"""
Get a MagicDotPathWithOp objects with a `NOT` operator.
Returns:
MagicDotPathWithOp.
"""
return self._get_one_operand_operator(OperatorType.INVERT)
def __pow__(
self,
other: _VALID_OPERAND_TYPE | BaseMagicDotPath,
) -> MagicDotPathWithOp:
"""
Get a MagicDotPathWithOp objects with a `^` operator.
Args:
other: The value at which the comparison is made.
Returns:
MagicDotPathWithOp.
Raises:
TypeOperatorError: Indirect raise by `_get_number_operator`
"""
return self._get_number_operator(other, OperatorType.POW)
def __rpow__(
self,
other: _VALID_OPERAND_TYPE | BaseMagicDotPath,
) -> MagicDotPathWithOp:
"""
Get a MagicDotPathWithOp objects with a `^` operator and the correct types.
This operator is useful when we make `8 ^ MagicDotPath`
and not `MagicDotPath ^ 8`.
Types are integer, float, Decimal, BaseMagicDotPath.
Args:
other: The value at which the comparison is made.
Returns:
MagicDotPathWithOp.
Raises:
TypeOperatorError: Indirect raise by `_get_number_operator`
"""
mdp_with_op = self._get_number_operator(self, OperatorType.POW)
mdp_with_op.operand_1 = other
return mdp_with_op
def __abs__(self) -> MagicDotPathWithOp:
"""
Get a MagicDotPathWithOp objects with a `abs()` operator.
Returns:
MagicDotPathWithOp.
"""
return self._get_one_operand_operator(OperatorType.ABS)
def __pos__(self) -> MagicDotPathWithOp:
"""
Get a MagicDotPathWithOp objects with a `+` operator.
Returns:
MagicDotPathWithOp.
"""
return self._get_one_operand_operator(OperatorType.POS)
def __neg__(self) -> MagicDotPathWithOp:
"""
Get a MagicDotPathWithOp objects with a `-` operator.
Returns:
MagicDotPathWithOp.
"""
return self._get_one_operand_operator(OperatorType.NEG)
def jsonb_path(self, as_str: bool) -> str:
"""
Get the corresponding jsonb path from a MagicDotPath.
Args:
as_str: Boolean to force the request with json in str.
Returns:
A path with the correct jsonb syntax.
Raises:
NotImplementedError: This function is just a wrapper for the subclasses.
"""
# No cover because is just a wrapper for subclasses.
raise NotImplementedError # pragma: no cover
def col_name(self) -> str:
"""
Get the corresponding column name form a MagicDotPath.
Returns:
A column name with the correct format.
Raises:
NotImplementedError: This function is just a wrapper for the subclasses.
"""
# No cover because is just a wrapper for subclasses.
raise NotImplementedError # pragma: no cover
__abs__(self)
special
Get a MagicDotPathWithOp objects with a abs()
operator.
Returns: |
|
---|
Source code in py_linq_sql/utils/classes/magicdotpath.py
def __abs__(self) -> MagicDotPathWithOp:
"""
Get a MagicDotPathWithOp objects with a `abs()` operator.
Returns:
MagicDotPathWithOp.
"""
return self._get_one_operand_operator(OperatorType.ABS)
__add__(self, other)
special
Get a MagicDotPathWithOp objects with a +
operator and the correct types.
Types are integer, float, Decimal, BaseMagicDotPath.
Parameters: |
|
---|
Returns: |
|
---|
Exceptions: |
|
---|
Source code in py_linq_sql/utils/classes/magicdotpath.py
def __add__(
self,
other: _VALID_OPERAND_TYPE | BaseMagicDotPath,
) -> MagicDotPathWithOp:
"""
Get a MagicDotPathWithOp objects with a `+` operator and the correct types.
Types are integer, float, Decimal, BaseMagicDotPath.
Args:
other: The value at which the comparison is made.
Returns:
MagicDotPathWithOp.
Raises:
TypeOperatorError: Indirect raise by `_get_number_operator`
"""
return self._get_number_operator(other, OperatorType.ADD)
__and__(self, other)
special
Get a MagicDotPathWithOp objects with a AND
operator and the correct types.
Types are integer, float, Decimal, BaseMagicDotPath.
Parameters: |
|
---|
Returns: |
|
---|
Exceptions: |
|
---|
Source code in py_linq_sql/utils/classes/magicdotpath.py
def __and__(
self,
other: _VALID_OPERAND_TYPE | BaseMagicDotPath,
) -> MagicDotPathWithOp:
"""
Get a MagicDotPathWithOp objects with a `AND` operator and the correct types.
Types are integer, float, Decimal, BaseMagicDotPath.
Args:
other: The value at which the comparison is made.
Returns:
MagicDotPathWithOp.
Raises:
TypeOperatorError: Indirect raise by `_get_generic_operator`
"""
return self._get_generic_operator(other, OperatorType.AND)
__eq__(self, other)
special
Get a MagicDotPathWithOp objects with a =
operator and the correct types.
Types are integer, float, Decimal, string, list, dict, BaseMagicDotPath.
Parameters: |
|
---|
Returns: |
|
---|
Exceptions: |
|
---|
Source code in py_linq_sql/utils/classes/magicdotpath.py
def __eq__( # type: ignore[override]
self,
other: _VALID_OPERAND_TYPE | BaseMagicDotPath,
) -> MagicDotPathWithOp:
"""
Get a MagicDotPathWithOp objects with a `=` operator and the correct types.
Types are integer, float, Decimal, string, list, dict, BaseMagicDotPath.
Args:
other: The value at which the comparison is made.
Returns:
MagicDotPathWithOp.
Raises:
TypeOperatorError: Indirect raise by `_get_generic_operator`
"""
return self._get_generic_operator(other, OperatorType.EQ)
__ge__(self, other)
special
Get a MagicDotPathWithOp objects with a >=
operator and the correct types.
Types are integer, float, Decimal or BaseMagicDotPath.
Parameters: |
|
---|
Returns: |
|
---|
Exceptions: |
|
---|
Source code in py_linq_sql/utils/classes/magicdotpath.py
def __ge__(
self,
other: _VALID_OPERAND_TYPE | BaseMagicDotPath,
) -> MagicDotPathWithOp:
"""
Get a MagicDotPathWithOp objects with a `>=` operator and the correct types.
Types are integer, float, Decimal or BaseMagicDotPath.
Args:
other: The value at which the comparison is made.
Returns:
MagicDotPathWithOp.
Raises:
TypeOperatorError: Indirect raise by `_get_number_operator`
"""
return self._get_number_operator(other, OperatorType.GE)
__getattr__(self, attribute_name)
special
Get attribute in a list for a MagicDotPath objects.
Parameters: |
|
---|
Returns: |
|
---|
Exceptions: |
|
---|
Source code in py_linq_sql/utils/classes/magicdotpath.py
def __getattr__(self, attribute_name: str) -> BaseMagicDotPath:
"""
Get attribute in a list for a MagicDotPath objects.
Args:
attribute_name: Names of all attributes.
Returns:
A MagicDotPath objects with attributes names in attributes list.
Raises:
NotImplementedError: This function is just a wrapper for the subclasses.
"""
# No cover because is just a wrapper for subclasses.
raise NotImplementedError # pragma: no cover
__getitem__(self, attribute_name)
special
Get items in a list for a MagicDotPath objects.
Parameters: |
|
---|
Returns: |
|
---|
Exceptions: |
|
---|
Source code in py_linq_sql/utils/classes/magicdotpath.py
def __getitem__(self, attribute_name: str) -> BaseMagicDotPath:
"""
Get items in a list for a MagicDotPath objects.
Args:
attribute_name: Names of all attributes.
Returns:
A MagicDotPath objects with attributes names in attributes list.
Raises:
NotImplementedError: This function is just a wrapper for the subclasses.
"""
# No cover because is just a wrapper for subclasses.
raise NotImplementedError # pragma: no cover
__gt__(self, other)
special
Get a MagicDotPathWithOp objects with a >
operator and the correct types.
Types are integer, float, Decimal or BaseMagicDotPath.
Parameters: |
|
---|
Returns: |
|
---|
Exceptions: |
|
---|
Source code in py_linq_sql/utils/classes/magicdotpath.py
def __gt__(
self,
other: _VALID_OPERAND_TYPE | BaseMagicDotPath,
) -> MagicDotPathWithOp:
"""
Get a MagicDotPathWithOp objects with a `>` operator and the correct types.
Types are integer, float, Decimal or BaseMagicDotPath.
Args:
other: The value at which the comparison is made.
Returns:
MagicDotPathWithOp.
Raises:
TypeOperatorError: Indirect raise by `_get_number_operator`
"""
return self._get_number_operator(other, OperatorType.GT)
__invert__(self)
special
Get a MagicDotPathWithOp objects with a NOT
operator.
Returns: |
|
---|
Source code in py_linq_sql/utils/classes/magicdotpath.py
def __invert__(self) -> MagicDotPathWithOp:
"""
Get a MagicDotPathWithOp objects with a `NOT` operator.
Returns:
MagicDotPathWithOp.
"""
return self._get_one_operand_operator(OperatorType.INVERT)
__le__(self, other)
special
Get a MagicDotPathWithOp objects with a <=
operator and the correct types.
Types are integer, float, Decimal or BaseMagicDotPath.
Parameters: |
|
---|
Returns: |
|
---|
Exceptions: |
|
---|
Source code in py_linq_sql/utils/classes/magicdotpath.py
def __le__(
self,
other: _VALID_OPERAND_TYPE | BaseMagicDotPath,
) -> MagicDotPathWithOp:
"""
Get a MagicDotPathWithOp objects with a `<=` operator and the correct types.
Types are integer, float, Decimal or BaseMagicDotPath.
Args:
other: The value at which the comparison is made.
Returns:
MagicDotPathWithOp.
Raises:
TypeOperatorError: Indirect raise by `_get_number_operator`
"""
return self._get_number_operator(other, OperatorType.LE)
__lt__(self, other)
special
Get a MagicDotPathWithOp objects with a <
operator and the correct types.
Types are integer, float, Decimal or BaseMagicDotPath.
Parameters: |
|
---|
Returns: |
|
---|
Exceptions: |
|
---|
Source code in py_linq_sql/utils/classes/magicdotpath.py
def __lt__(
self,
other: _VALID_OPERAND_TYPE | BaseMagicDotPath,
) -> MagicDotPathWithOp:
"""
Get a MagicDotPathWithOp objects with a `<` operator and the correct types.
Types are integer, float, Decimal or BaseMagicDotPath.
Args:
other: The value at which the comparison is made.
Returns:
MagicDotPathWithOp.
Raises:
TypeOperatorError: Indirect raise by `_get_number_operator`
"""
return self._get_number_operator(other, OperatorType.LT)
__mod__(self, other)
special
Get a MagicDotPathWithOp objects with a %
operator and the correct types.
Types are integer, float, Decimal, BaseMagicDotPath.
Parameters: |
|
---|
Returns: |
|
---|
Exceptions: |
|
---|
Source code in py_linq_sql/utils/classes/magicdotpath.py
def __mod__(
self,
other: _VALID_OPERAND_TYPE | BaseMagicDotPath,
) -> MagicDotPathWithOp:
"""
Get a MagicDotPathWithOp objects with a `%` operator and the correct types.
Types are integer, float, Decimal, BaseMagicDotPath.
Args:
other: The value at which the comparison is made.
Returns:
MagicDotPathWithOp.
Raises:
TypeOperatorError: Indirect raise by `_get_number_operator`
"""
return self._get_number_operator(other, OperatorType.MOD)
__mul__(self, other)
special
Get a MagicDotPathWithOp objects with a *
operator and the correct types.
Types are integer, float, Decimal, BaseMagicDotPath.
Parameters: |
|
---|
Returns: |
|
---|
Exceptions: |
|
---|
Source code in py_linq_sql/utils/classes/magicdotpath.py
def __mul__(
self,
other: _VALID_OPERAND_TYPE | BaseMagicDotPath,
) -> MagicDotPathWithOp:
"""
Get a MagicDotPathWithOp objects with a `*` operator and the correct types.
Types are integer, float, Decimal, BaseMagicDotPath.
Args:
other: The value at which the comparison is made.
Returns:
MagicDotPathWithOp.
Raises:
TypeOperatorError: Indirect raise by `_get_number_operator`
"""
return self._get_number_operator(other, OperatorType.MUL)
__ne__(self, other)
special
Get a MagicDotPathWithOp objects with a <>
operator and the correct types.
Types are integer, float, Decimal, string, list, dict, BaseMagicDotPath.
Parameters: |
|
---|
Returns: |
|
---|
Exceptions: |
|
---|
Source code in py_linq_sql/utils/classes/magicdotpath.py
def __ne__( # type: ignore[override]
self,
other: _VALID_OPERAND_TYPE | BaseMagicDotPath,
) -> MagicDotPathWithOp:
"""
Get a MagicDotPathWithOp objects with a `<>` operator and the correct types.
Types are integer, float, Decimal, string, list, dict, BaseMagicDotPath.
Args:
other: The value at which the comparison is made.
Returns:
MagicDotPathWithOp.
Raises:
TypeOperatorError: Indirect raise by `_get_generic_operator`
"""
return self._get_generic_operator(other, OperatorType.NE)
__neg__(self)
special
Get a MagicDotPathWithOp objects with a -
operator.
Returns: |
|
---|
Source code in py_linq_sql/utils/classes/magicdotpath.py
def __neg__(self) -> MagicDotPathWithOp:
"""
Get a MagicDotPathWithOp objects with a `-` operator.
Returns:
MagicDotPathWithOp.
"""
return self._get_one_operand_operator(OperatorType.NEG)
__or__(self, other)
special
Get a MagicDotPathWithOp objects with a OR
operator and the correct types.
Types are integer, float, Decimal, BaseMagicDotPath.
Parameters: |
|
---|
Returns: |
|
---|
Exceptions: |
|
---|
Source code in py_linq_sql/utils/classes/magicdotpath.py
def __or__(
self,
other: _VALID_OPERAND_TYPE | BaseMagicDotPath,
) -> MagicDotPathWithOp:
"""
Get a MagicDotPathWithOp objects with a `OR` operator and the correct types.
Types are integer, float, Decimal, BaseMagicDotPath.
Args:
other: The value at which the comparison is made.
Returns:
MagicDotPathWithOp.
Raises:
TypeOperatorError: Indirect raise by `_get_generic_operator`
"""
return self._get_generic_operator(other, OperatorType.OR)
__pos__(self)
special
Get a MagicDotPathWithOp objects with a +
operator.
Returns: |
|
---|
Source code in py_linq_sql/utils/classes/magicdotpath.py
def __pos__(self) -> MagicDotPathWithOp:
"""
Get a MagicDotPathWithOp objects with a `+` operator.
Returns:
MagicDotPathWithOp.
"""
return self._get_one_operand_operator(OperatorType.POS)
__pow__(self, other)
special
Get a MagicDotPathWithOp objects with a ^
operator.
Parameters: |
|
---|
Returns: |
|
---|
Exceptions: |
|
---|
Source code in py_linq_sql/utils/classes/magicdotpath.py
def __pow__(
self,
other: _VALID_OPERAND_TYPE | BaseMagicDotPath,
) -> MagicDotPathWithOp:
"""
Get a MagicDotPathWithOp objects with a `^` operator.
Args:
other: The value at which the comparison is made.
Returns:
MagicDotPathWithOp.
Raises:
TypeOperatorError: Indirect raise by `_get_number_operator`
"""
return self._get_number_operator(other, OperatorType.POW)
__radd__(self, other)
special
Get a MagicDotPathWithOp objects with a +
operator and the correct types.
This operator is useful when we make 8 + MagicDotPath
and not MagicDotPath + 8
.
Types are integer, float, Decimal, BaseMagicDotPath.
Parameters: |
|
---|
Returns: |
|
---|
Exceptions: |
|
---|
Source code in py_linq_sql/utils/classes/magicdotpath.py
def __radd__(
self,
other: _VALID_OPERAND_TYPE | BaseMagicDotPath,
) -> MagicDotPathWithOp:
"""
Get a MagicDotPathWithOp objects with a `+` operator and the correct types.
This operator is useful when we make `8 + MagicDotPath`
and not `MagicDotPath + 8`.
Types are integer, float, Decimal, BaseMagicDotPath.
Args:
other: The value at which the comparison is made.
Returns:
MagicDotPathWithOp.
Raises:
TypeOperatorError: Indirect raise by `_get_number_operator`
"""
return self.__add__(other)
__rand__(self, other)
special
Get a MagicDotPathWithOp objects with a AND
operator and the correct types.
This operator is useful when we make 8 & MagicDotPath
and not MagicDotPath & 8
.
Types are integer, float, Decimal, BaseMagicDotPath.
Parameters: |
|
---|
Returns: |
|
---|
Exceptions: |
|
---|
Source code in py_linq_sql/utils/classes/magicdotpath.py
def __rand__(
self,
other: _VALID_OPERAND_TYPE | BaseMagicDotPath,
) -> MagicDotPathWithOp:
"""
Get a MagicDotPathWithOp objects with a `AND` operator and the correct types.
This operator is useful when we make `8 & MagicDotPath`
and not `MagicDotPath & 8`.
Types are integer, float, Decimal, BaseMagicDotPath.
Args:
other: The value at which the comparison is made.
Returns:
MagicDotPathWithOp.
Raises:
TypeOperatorError: Indirect raise by `_get_generic_operator`
"""
return self.__and__(other)
__rmod__(self, other)
special
Get a MagicDotPathWithOp objects with a %
operator and the correct types.
This operator is useful when we make 8 % MagicDotPath
and not MagicDotPath % 8
.
Types are integer, float, Decimal, BaseMagicDotPath.
Parameters: |
|
---|
Returns: |
|
---|
Exceptions: |
|
---|
Source code in py_linq_sql/utils/classes/magicdotpath.py
def __rmod__(
self,
other: _VALID_OPERAND_TYPE | BaseMagicDotPath,
) -> MagicDotPathWithOp:
"""
Get a MagicDotPathWithOp objects with a `%` operator and the correct types.
This operator is useful when we make `8 % MagicDotPath`
and not `MagicDotPath % 8`.
Types are integer, float, Decimal, BaseMagicDotPath.
Args:
other: The value at which the comparison is made.
Returns:
MagicDotPathWithOp.
Raises:
TypeOperatorError: Indirect raise by `_get_number_operator`
"""
mdp_with_op = self._get_number_operator(self, OperatorType.MOD)
mdp_with_op.operand_1 = other
return mdp_with_op
__rmul__(self, other)
special
Get a MagicDotPathWithOp objects with a *
operator and the correct types.
This operator is useful when we make 8 * MagicDotPath
and not MagicDotPath * 8
.
Types are integer, float, Decimal, BaseMagicDotPath.
Parameters: |
|
---|
Returns: |
|
---|
Exceptions: |
|
---|
Source code in py_linq_sql/utils/classes/magicdotpath.py
def __rmul__(
self,
other: _VALID_OPERAND_TYPE | BaseMagicDotPath,
) -> MagicDotPathWithOp:
"""
Get a MagicDotPathWithOp objects with a `*` operator and the correct types.
This operator is useful when we make `8 * MagicDotPath`
and not `MagicDotPath * 8`.
Types are integer, float, Decimal, BaseMagicDotPath.
Args:
other: The value at which the comparison is made.
Returns:
MagicDotPathWithOp.
Raises:
TypeOperatorError: Indirect raise by `_get_number_operator`
"""
return self.__mul__(other)
__ror__(self, other)
special
Get a MagicDotPathWithOp objects with a OR
operator and the correct types.
This operator is useful when we make 8 | MagicDotPath
and not MagicDotPath | 8
.
Types are integer, float, Decimal, BaseMagicDotPath.
Parameters: |
|
---|
Returns: |
|
---|
Exceptions: |
|
---|
Source code in py_linq_sql/utils/classes/magicdotpath.py
def __ror__(
self,
other: _VALID_OPERAND_TYPE | BaseMagicDotPath,
) -> MagicDotPathWithOp:
"""
Get a MagicDotPathWithOp objects with a `OR` operator and the correct types.
This operator is useful when we make `8 | MagicDotPath`
and not `MagicDotPath | 8`.
Types are integer, float, Decimal, BaseMagicDotPath.
Args:
other: The value at which the comparison is made.
Returns:
MagicDotPathWithOp.
Raises:
TypeOperatorError: Indirect raise by `_get_generic_operator`
"""
return self.__or__(other)
__rpow__(self, other)
special
Get a MagicDotPathWithOp objects with a ^
operator and the correct types.
This operator is useful when we make 8 ^ MagicDotPath
and not MagicDotPath ^ 8
.
Types are integer, float, Decimal, BaseMagicDotPath.
Parameters: |
|
---|
Returns: |
|
---|
Exceptions: |
|
---|
Source code in py_linq_sql/utils/classes/magicdotpath.py
def __rpow__(
self,
other: _VALID_OPERAND_TYPE | BaseMagicDotPath,
) -> MagicDotPathWithOp:
"""
Get a MagicDotPathWithOp objects with a `^` operator and the correct types.
This operator is useful when we make `8 ^ MagicDotPath`
and not `MagicDotPath ^ 8`.
Types are integer, float, Decimal, BaseMagicDotPath.
Args:
other: The value at which the comparison is made.
Returns:
MagicDotPathWithOp.
Raises:
TypeOperatorError: Indirect raise by `_get_number_operator`
"""
mdp_with_op = self._get_number_operator(self, OperatorType.POW)
mdp_with_op.operand_1 = other
return mdp_with_op
__rsub__(self, other)
special
Get a MagicDotPathWithOp objects with a -
operator and the correct types.
This operator is useful when we make 8 - MagicDotPath
and not MagicDotPath - 8
.
Types are integer, float, Decimal, BaseMagicDotPath.
Parameters: |
|
---|
Returns: |
|
---|
Exceptions: |
|
---|
Source code in py_linq_sql/utils/classes/magicdotpath.py
def __rsub__(
self,
other: _VALID_OPERAND_TYPE | BaseMagicDotPath,
) -> MagicDotPathWithOp:
"""
Get a MagicDotPathWithOp objects with a `-` operator and the correct types.
This operator is useful when we make `8 - MagicDotPath`
and not `MagicDotPath - 8`.
Types are integer, float, Decimal, BaseMagicDotPath.
Args:
other: The value at which the comparison is made.
Returns:
MagicDotPathWithOp.
Raises:
TypeOperatorError: Indirect raise by `_get_number_operator`
"""
mdp_with_op = self._get_number_operator(self, OperatorType.SUB)
mdp_with_op.operand_1 = other
return mdp_with_op
__rtruediv__(self, other)
special
Get a MagicDotPathWithOp objects with a /
operator and the correct types.
This operator is useful when we make 8 / MagicDotPath
and not MagicDotPath / 8
.
Types are integer, float, Decimal, BaseMagicDotPath.
Parameters: |
|
---|
Returns: |
|
---|
Exceptions: |
|
---|
Source code in py_linq_sql/utils/classes/magicdotpath.py
def __rtruediv__(
self,
other: _VALID_OPERAND_TYPE | BaseMagicDotPath,
) -> MagicDotPathWithOp:
"""
Get a MagicDotPathWithOp objects with a `/` operator and the correct types.
This operator is useful when we make `8 / MagicDotPath`
and not `MagicDotPath / 8`.
Types are integer, float, Decimal, BaseMagicDotPath.
Args:
other: The value at which the comparison is made.
Returns:
MagicDotPathWithOp.
Raises:
TypeOperatorError: Indirect raise by `_get_number_operator`
"""
mdp_with_op = self._get_number_operator(self, OperatorType.TRUEDIV)
mdp_with_op.operand_1 = other
return mdp_with_op
__sub__(self, other)
special
Get a MagicDotPathWithOp objects with a -
operator and the correct types.
Types are integer, float, Decimal, BaseMagicDotPath.
Parameters: |
|
---|
Returns: |
|
---|
Exceptions: |
|
---|
Source code in py_linq_sql/utils/classes/magicdotpath.py
def __sub__(
self,
other: _VALID_OPERAND_TYPE | BaseMagicDotPath,
) -> MagicDotPathWithOp:
"""
Get a MagicDotPathWithOp objects with a `-` operator and the correct types.
Types are integer, float, Decimal, BaseMagicDotPath.
Args:
other: The value at which the comparison is made.
Returns:
MagicDotPathWithOp.
Raises:
TypeOperatorError: Indirect raise by `_get_number_operator`
"""
return self._get_number_operator(other, OperatorType.SUB)
__truediv__(self, other)
special
Get a MagicDotPathWithOp objects with a /
operator and the correct types.
Types are integer, float, Decimal, BaseMagicDotPath.
Parameters: |
|
---|
Returns: |
|
---|
Exceptions: |
|
---|
Source code in py_linq_sql/utils/classes/magicdotpath.py
def __truediv__(
self,
other: _VALID_OPERAND_TYPE | BaseMagicDotPath,
) -> MagicDotPathWithOp:
"""
Get a MagicDotPathWithOp objects with a `/` operator and the correct types.
Types are integer, float, Decimal, BaseMagicDotPath.
Args:
other: The value at which the comparison is made.
Returns:
MagicDotPathWithOp.
Raises:
TypeOperatorError: Indirect raise by `_get_number_operator`
"""
return self._get_number_operator(other, OperatorType.TRUEDIV)
col_name(self)
Get the corresponding column name form a MagicDotPath.
Returns: |
|
---|
Exceptions: |
|
---|
Source code in py_linq_sql/utils/classes/magicdotpath.py
def col_name(self) -> str:
"""
Get the corresponding column name form a MagicDotPath.
Returns:
A column name with the correct format.
Raises:
NotImplementedError: This function is just a wrapper for the subclasses.
"""
# No cover because is just a wrapper for subclasses.
raise NotImplementedError # pragma: no cover
jsonb_path(self, as_str)
Get the corresponding jsonb path from a MagicDotPath.
Parameters: |
|
---|
Returns: |
|
---|
Exceptions: |
|
---|
Source code in py_linq_sql/utils/classes/magicdotpath.py
def jsonb_path(self, as_str: bool) -> str:
"""
Get the corresponding jsonb path from a MagicDotPath.
Args:
as_str: Boolean to force the request with json in str.
Returns:
A path with the correct jsonb syntax.
Raises:
NotImplementedError: This function is just a wrapper for the subclasses.
"""
# No cover because is just a wrapper for subclasses.
raise NotImplementedError # pragma: no cover
MagicDotPath
Magical object that can have any attribute.
Inspired by The very useful DotMap module this
object allow to write predicate in the lambda function used in SQL clause like
.select()
or .join()
. This is very useful to express SQL constraint.
See the LINQ documentation for more in depth explanation.
Attributes:
Name | Type | Description |
---|---|---|
connection |
Connection |
Connection to a db. Useful for safe. |
attributes |
list[str] |
Attributes for build the jsonb path. |
with_table |
str | None |
Table on which we want to get paths. |
column |
str | None |
Column on which we will want to make the request. |
Source code in py_linq_sql/utils/classes/magicdotpath.py
@dataclass(eq=False)
class MagicDotPath(BaseMagicDotPath):
"""
Magical object that can have any attribute.
Inspired by The very useful [DotMap module](https://github.com/drgrib/dotmap) this
object allow to write predicate in the lambda function used in SQL clause like
`.select()` or `.join()`. This is very useful to express SQL constraint.
See the LINQ documentation for more in depth explanation.
Attributes:
connection: Connection to a db. Useful for safe.
attributes: Attributes for build the jsonb path.
with_table: Table on which we want to get paths.
column: Column on which we will want to make the request.
"""
connection: Connection
attributes: list[str] = field(default_factory=list)
with_table: str | None = None
column: str | None = None
def __getattr__(self, attribute_name: str) -> MagicDotPath:
"""
Get attribute in a list for a MagicDotPath objects.
Args:
attribute_name: Names of all attributes.
Returns:
A MagicDotPath objects with attributes names in attributes list.
"""
return MagicDotPath(
self.connection,
attributes=self.attributes + [f"'{attribute_name}'"],
with_table=self.with_table,
)
def __getitem__(self, attribute_name: str) -> MagicDotPath:
"""
Get attr from the dict syntax.
Args:
attribute_name: Names of all attributes.
Returns:
A MagicDotPath objects with attributes names in attributes list.
"""
return self.__getattr__(attribute_name)
def safe(self, name: str) -> str:
"""
Secure a column or a table for a request.
Args:
name: Name of the column or table we want to secure.
Returns:
Name but verified by psycopg.sql.Identifier.
Raises:
psycopg.Error: Indirect raise by `sql.Identifier` or `as_string`.
"""
return sql.Identifier(name).as_string(self.connection)
def format_table_with_dot(self) -> str:
"""
Get the table with dot in good format.
Returns:
The table in the good psql format.
"""
return ".".join(
[f"{tab}" for tab in self.with_table.split(".")] # type: ignore[union-attr]
)
def jsonb_path(self, as_str: bool) -> str:
"""
Get the corresponding jsonb path from a MagicDotPath.
Args:
as_str: Boolean to force the request with json in str.
Returns:
A path with the correct jsonb syntax.
Raises:
psycopg.Error: Indirect raise by `safe`.
"""
self.column = self.safe(self.attributes[0][1:-1])
res = ""
if self.with_table:
table = (
self.format_table_with_dot()
if "." in self.with_table
else f"{self.with_table}"
)
res = f"{res}{table}."
if len(self.attributes) == 1:
res = f"{res}{self.column}"
else:
if as_str:
path_item = self.attributes[1:-1].copy()
path_item.insert(0, self.column)
res = f'{res}{"->".join(path_item)}->>{self.attributes[-1]}'
else:
res = f'{res}{self.column}->{"->".join(self.attributes[1:])}'
return res
def col_name(self) -> str:
"""
Get the corresponding column name form a MagicDotPath.
Returns:
A column name with the correct format.
"""
result = []
if self.with_table:
table = (
self.format_table_with_dot()
if "." in self.with_table
else self.with_table
)
result.append(table[1:-1])
for att in self.attributes:
result.append(att[1:-1])
return "_".join(result)
__getattr__(self, attribute_name)
special
Get attribute in a list for a MagicDotPath objects.
Parameters: |
|
---|
Returns: |
|
---|
Source code in py_linq_sql/utils/classes/magicdotpath.py
def __getattr__(self, attribute_name: str) -> MagicDotPath:
"""
Get attribute in a list for a MagicDotPath objects.
Args:
attribute_name: Names of all attributes.
Returns:
A MagicDotPath objects with attributes names in attributes list.
"""
return MagicDotPath(
self.connection,
attributes=self.attributes + [f"'{attribute_name}'"],
with_table=self.with_table,
)
__getitem__(self, attribute_name)
special
Get attr from the dict syntax.
Parameters: |
|
---|
Returns: |
|
---|
Source code in py_linq_sql/utils/classes/magicdotpath.py
def __getitem__(self, attribute_name: str) -> MagicDotPath:
"""
Get attr from the dict syntax.
Args:
attribute_name: Names of all attributes.
Returns:
A MagicDotPath objects with attributes names in attributes list.
"""
return self.__getattr__(attribute_name)
col_name(self)
Get the corresponding column name form a MagicDotPath.
Returns: |
|
---|
Source code in py_linq_sql/utils/classes/magicdotpath.py
def col_name(self) -> str:
"""
Get the corresponding column name form a MagicDotPath.
Returns:
A column name with the correct format.
"""
result = []
if self.with_table:
table = (
self.format_table_with_dot()
if "." in self.with_table
else self.with_table
)
result.append(table[1:-1])
for att in self.attributes:
result.append(att[1:-1])
return "_".join(result)
format_table_with_dot(self)
Get the table with dot in good format.
Returns: |
|
---|
Source code in py_linq_sql/utils/classes/magicdotpath.py
def format_table_with_dot(self) -> str:
"""
Get the table with dot in good format.
Returns:
The table in the good psql format.
"""
return ".".join(
[f"{tab}" for tab in self.with_table.split(".")] # type: ignore[union-attr]
)
jsonb_path(self, as_str)
Get the corresponding jsonb path from a MagicDotPath.
Parameters: |
|
---|
Returns: |
|
---|
Exceptions: |
|
---|
Source code in py_linq_sql/utils/classes/magicdotpath.py
def jsonb_path(self, as_str: bool) -> str:
"""
Get the corresponding jsonb path from a MagicDotPath.
Args:
as_str: Boolean to force the request with json in str.
Returns:
A path with the correct jsonb syntax.
Raises:
psycopg.Error: Indirect raise by `safe`.
"""
self.column = self.safe(self.attributes[0][1:-1])
res = ""
if self.with_table:
table = (
self.format_table_with_dot()
if "." in self.with_table
else f"{self.with_table}"
)
res = f"{res}{table}."
if len(self.attributes) == 1:
res = f"{res}{self.column}"
else:
if as_str:
path_item = self.attributes[1:-1].copy()
path_item.insert(0, self.column)
res = f'{res}{"->".join(path_item)}->>{self.attributes[-1]}'
else:
res = f'{res}{self.column}->{"->".join(self.attributes[1:])}'
return res
safe(self, name)
Secure a column or a table for a request.
Parameters: |
|
---|
Returns: |
|
---|
Exceptions: |
|
---|
Source code in py_linq_sql/utils/classes/magicdotpath.py
def safe(self, name: str) -> str:
"""
Secure a column or a table for a request.
Args:
name: Name of the column or table we want to secure.
Returns:
Name but verified by psycopg.sql.Identifier.
Raises:
psycopg.Error: Indirect raise by `sql.Identifier` or `as_string`.
"""
return sql.Identifier(name).as_string(self.connection)
MagicDotPathWithOp
Magical object that can have any attribute and can be subject to many comparison.
This class inherited from MagicDotPath.
Inspired by The very useful DotMap module this
object allow to write predicate in the lambda function used in SQL clause like
.select()
or .join()
. This is very useful to express SQL constraint.
See the LINQ documentation for more in depth explanation.
Attributes:
Name | Type | Description |
---|---|---|
operand_1 |
py_linq_sql.utils.classes.magicdotpath.BaseMagicDotPath | float | int | str | list | dict | decimal.Decimal | bool |
First operand for the operation. |
operand_2 |
py_linq_sql.utils.classes.magicdotpath.BaseMagicDotPath | float | int | str | list | dict | decimal.Decimal | bool | None |
Second operand for the operation. |
operator |
py_linq_sql.utils.classes.op_and_func_of_mdp.HyperBFuncType | py_linq_sql.utils.classes.op_and_func_of_mdp.MathFunctType | py_linq_sql.utils.classes.op_and_func_of_mdp.OperatorType | py_linq_sql.utils.classes.op_and_func_of_mdp.TrigoFunctType | py_linq_sql.utils.classes.op_and_func_of_mdp.StrFunctType |
Operator for comparison. |
my_type |
Type[Any] |
Type of other. |
Source code in py_linq_sql/utils/classes/magicdotpath.py
@dataclass(eq=False)
class MagicDotPathWithOp(BaseMagicDotPath):
"""
Magical object that can have any attribute and can be subject to many comparison.
This class inherited from MagicDotPath.
Inspired by The very useful [DotMap module](https://github.com/drgrib/dotmap) this
object allow to write predicate in the lambda function used in SQL clause like
`.select()` or `.join()`. This is very useful to express SQL constraint.
See the LINQ documentation for more in depth explanation.
Attributes:
operand_1: First operand for the operation.
operand_2: Second operand for the operation.
operator: Operator for comparison.
my_type: Type of other.
"""
operand_1: BaseMagicDotPath | _VALID_OPERAND_TYPE
operator: _OPERATORTYPE
operand_2: BaseMagicDotPath | _VALID_OPERAND_TYPE | None
my_type: Type[Any]
def jsonb_path(self, as_str: bool) -> str:
"""
Get the corresponding jsonb path from a MagicDotPathWithOp.
Args:
as_str: Boolean to force the request with json in str.
Returns:
A path with the correct jsonb syntax.
Raises:
psycopg.Error: Indirect raise by `MagicDotPath.safe`.
TypeOperatorError: Indirect raise by `json_path_hyper`, `json_path_math`,
`json_path_ope` or `json_path_trigo`.
"""
as_str = self.my_type == str
path_op = DotMap()
match self.operand_1:
case MagicDotPath() | MagicDotPathWithOp():
path_op.op1 = self.operand_1.jsonb_path(as_str)
case _:
path_op.op1 = str(self.operand_1)
match self.operand_2:
case MagicDotPath() | MagicDotPathWithOp():
path_op.op2 = self.operand_2.jsonb_path(as_str)
case dict():
operand_2_in_good_format = json.dumps(self.operand_2)
path_op.op2 = f"'{operand_2_in_good_format}'::jsonb"
case list():
if self.operator in [OperatorType.IN]:
path_op.op2 = (
f"""({", ".join([f"'{val}'" for val in self.operand_2])})"""
)
else:
path_op.op2 = f"'{self.operand_2}'::jsonb"
case _:
path_op.op2 = str(self.operand_2)
match self.my_type:
case type() if self.my_type == bool:
result = (
f"CAST({path_op.op1} AS boolean) {self.operator.psql} {path_op.op2}"
)
case type() if self.operator in [
OperatorType.AND,
OperatorType.OR,
OperatorType.INVERT,
OperatorType.ABS,
OperatorType.POS,
OperatorType.NEG,
OperatorType.IN,
]:
result = json_path_ope(path_op, self.operator)
case type() if isinstance(self.operator, HyperBFuncType):
result = json_path_hyper(path_op, self.operator)
case type() if isinstance(self.operator, MathFunctType):
result = json_path_math(path_op, self.operator)
case type() if isinstance(self.operator, TrigoFunctType):
result = json_path_trigo(path_op, self.operator)
case type() if isinstance(self.operator, StrFunctType):
result = json_path_str(path_op, self.operator)
case type() if self.my_type == BaseMagicDotPath:
result = (
f"CAST({path_op.op1} AS decimal) "
f"{self.operator.psql} "
f"CAST({path_op.op2} AS decimal)"
)
case type() if self.my_type in (int, float, Decimal):
result = (
f"CAST({path_op.op1} AS decimal) {self.operator.psql} {path_op.op2}"
)
case type() if self.my_type == str:
result = f"{path_op.op1} {self.operator.psql} '{path_op.op2}'"
case _:
result = f"{path_op.op1} {self.operator.psql} {path_op.op2}"
return result
def col_name(self) -> str:
"""
Get the corresponding column name form a MagicDotPath.
Returns:
A column name with the correct format.
Raises:
TypeOperatorError: Indirect raise by `col_name_hyper`, `col_name_math`,
`col_name_ope` or `col_name_trigo`.
"""
name_op = DotMap()
match self.operand_1:
case BaseMagicDotPath():
name_op.op1 = self.operand_1.col_name()
case _:
name_op.op1 = str(self.operand_1)
match self.operand_2:
case None:
name_op.op2 = None
case BaseMagicDotPath():
name_op.op2 = self.operand_2.col_name()
case float() | int() | Decimal():
name_op.op2 = _clean_number_str(str(self.operand_2))
# TODO: test this
case list():
name_op.op2 = f"""({','.join([str(val) for val in self.operand_2])})"""
operator = self.operator
match operator:
case operator if operator in HyperBFuncType:
result = col_name_hyper(name_op, operator)
case operator if operator in MathFunctType:
result = col_name_math(name_op, operator)
case operator if operator in OperatorType:
result = col_name_ope(name_op, operator)
case operator if operator in TrigoFunctType:
result = col_name_trigo(name_op, operator)
case operator if operator in StrFunctType:
result = col_name_str(name_op, operator)
# No cover because is just an other security,
# but we can't go in this case thanks to Enum.
case _: # pragma: no cover
result = f"{name_op.op1}_???_{name_op.op2}"
return f"{result}"
col_name(self)
Get the corresponding column name form a MagicDotPath.
Returns: |
|
---|
Exceptions: |
|
---|
Source code in py_linq_sql/utils/classes/magicdotpath.py
def col_name(self) -> str:
"""
Get the corresponding column name form a MagicDotPath.
Returns:
A column name with the correct format.
Raises:
TypeOperatorError: Indirect raise by `col_name_hyper`, `col_name_math`,
`col_name_ope` or `col_name_trigo`.
"""
name_op = DotMap()
match self.operand_1:
case BaseMagicDotPath():
name_op.op1 = self.operand_1.col_name()
case _:
name_op.op1 = str(self.operand_1)
match self.operand_2:
case None:
name_op.op2 = None
case BaseMagicDotPath():
name_op.op2 = self.operand_2.col_name()
case float() | int() | Decimal():
name_op.op2 = _clean_number_str(str(self.operand_2))
# TODO: test this
case list():
name_op.op2 = f"""({','.join([str(val) for val in self.operand_2])})"""
operator = self.operator
match operator:
case operator if operator in HyperBFuncType:
result = col_name_hyper(name_op, operator)
case operator if operator in MathFunctType:
result = col_name_math(name_op, operator)
case operator if operator in OperatorType:
result = col_name_ope(name_op, operator)
case operator if operator in TrigoFunctType:
result = col_name_trigo(name_op, operator)
case operator if operator in StrFunctType:
result = col_name_str(name_op, operator)
# No cover because is just an other security,
# but we can't go in this case thanks to Enum.
case _: # pragma: no cover
result = f"{name_op.op1}_???_{name_op.op2}"
return f"{result}"
jsonb_path(self, as_str)
Get the corresponding jsonb path from a MagicDotPathWithOp.
Parameters: |
|
---|
Returns: |
|
---|
Exceptions: |
|
---|
Source code in py_linq_sql/utils/classes/magicdotpath.py
def jsonb_path(self, as_str: bool) -> str:
"""
Get the corresponding jsonb path from a MagicDotPathWithOp.
Args:
as_str: Boolean to force the request with json in str.
Returns:
A path with the correct jsonb syntax.
Raises:
psycopg.Error: Indirect raise by `MagicDotPath.safe`.
TypeOperatorError: Indirect raise by `json_path_hyper`, `json_path_math`,
`json_path_ope` or `json_path_trigo`.
"""
as_str = self.my_type == str
path_op = DotMap()
match self.operand_1:
case MagicDotPath() | MagicDotPathWithOp():
path_op.op1 = self.operand_1.jsonb_path(as_str)
case _:
path_op.op1 = str(self.operand_1)
match self.operand_2:
case MagicDotPath() | MagicDotPathWithOp():
path_op.op2 = self.operand_2.jsonb_path(as_str)
case dict():
operand_2_in_good_format = json.dumps(self.operand_2)
path_op.op2 = f"'{operand_2_in_good_format}'::jsonb"
case list():
if self.operator in [OperatorType.IN]:
path_op.op2 = (
f"""({", ".join([f"'{val}'" for val in self.operand_2])})"""
)
else:
path_op.op2 = f"'{self.operand_2}'::jsonb"
case _:
path_op.op2 = str(self.operand_2)
match self.my_type:
case type() if self.my_type == bool:
result = (
f"CAST({path_op.op1} AS boolean) {self.operator.psql} {path_op.op2}"
)
case type() if self.operator in [
OperatorType.AND,
OperatorType.OR,
OperatorType.INVERT,
OperatorType.ABS,
OperatorType.POS,
OperatorType.NEG,
OperatorType.IN,
]:
result = json_path_ope(path_op, self.operator)
case type() if isinstance(self.operator, HyperBFuncType):
result = json_path_hyper(path_op, self.operator)
case type() if isinstance(self.operator, MathFunctType):
result = json_path_math(path_op, self.operator)
case type() if isinstance(self.operator, TrigoFunctType):
result = json_path_trigo(path_op, self.operator)
case type() if isinstance(self.operator, StrFunctType):
result = json_path_str(path_op, self.operator)
case type() if self.my_type == BaseMagicDotPath:
result = (
f"CAST({path_op.op1} AS decimal) "
f"{self.operator.psql} "
f"CAST({path_op.op2} AS decimal)"
)
case type() if self.my_type in (int, float, Decimal):
result = (
f"CAST({path_op.op1} AS decimal) {self.operator.psql} {path_op.op2}"
)
case type() if self.my_type == str:
result = f"{path_op.op1} {self.operator.psql} '{path_op.op2}'"
case _:
result = f"{path_op.op1} {self.operator.psql} {path_op.op2}"
return result
MagicDotPathAggregate
MagicDotPath for aggregation.
Attributes:
Name | Type | Description |
---|---|---|
mdp |
MagicDotPath |
A MagicDotPath. |
operand |
py_linq_sql.utils.classes.magicdotpath.AggregateType | str |
Aggregation type. |
separator |
str | None |
Separator for string_agg aggregate. In other case its None. |
Source code in py_linq_sql/utils/classes/magicdotpath.py
@dataclass
class MagicDotPathAggregate:
"""
MagicDotPath for aggregation.
Attributes:
mdp: A MagicDotPath.
operand: Aggregation type.
separator: Separator for string_agg aggregate. In other case its None.
"""
mdp: MagicDotPath
operand: AggregateType | str # HACK: It's just the time that mypy supports the
# Strenum. https://github.com/python/mypy/issues
cast_type: type
separator: str | None = None
SQLEnumerableData class
SQLEnumerable with only data for building.
Attributes:
Name | Type | Description |
---|---|---|
connection |
Connection) |
Connection on which we want to execute the request. |
flags |
Flags |
All flags use to know the statement of the request. |
cmd |
List[Command] |
Commands we want to execute. |
table |
str | SQLEnumerable |
Table on which we want to execute the request. |
length |
int | None |
Length of the result of the request if we need else None. |
Source code in py_linq_sql/utils/classes/other_classes.py
@dataclass
class SQLEnumerableData:
"""
SQLEnumerable with only data for building.
Attributes:
connection (Connection) : Connection on which we want to execute the request.
flags (Flags): All flags use to know the statement of the request.
cmd (List[Command]): Commands we want to execute.
table (str | SQLEnumerable): Table on which we want to execute the request.
length (int | None): Length of the result of the request if we need else None.
"""
connection: Connection
flags: Flags
cmd: list[Command]
table: str | Any # it's an SQLEnumerable
length: int | None
Configurations
Functions to manage the configuration of py-linq-sql.
is_read_only()
Verify if the library configuration is read only or not.
Returns: |
|
---|
Source code in py_linq_sql/config/config.py
def is_read_only() -> bool:
"""
Verify if the library configuration is read only or not.
Returns:
True if we want denied write to database, False otherwise.
"""
config = _get_config()
return cast(bool, config["readonly"])
is_valid_table_name_with_white_and_black_list(table_name)
Verify if the table name is not in the black list and in the white list.
If no (or empty) white and/or black list we treat the specific cases according to: black list white list.
See the documentation to learn more proposed specific cases.
Parameters: |
|
---|
Returns: |
|
---|
Source code in py_linq_sql/config/config.py
def is_valid_table_name_with_white_and_black_list( # type: ignore[return]
table_name: str,
) -> bool:
"""
Verify if the table name is not in the black list and in the white list.
If no (or empty) white and/or black list we treat the specific cases according to:
black list white list.
See the documentation to learn more proposed specific cases.
Args:
table_name: The name of the table to validate.
Returns:
True is the table name is valid, False otherwise.
"""
config = _get_config()
w_list = (
None
if config["whitelist"] is None
else set(cast(Iterable, config["whitelist"]))
)
b_list = (
None
if config["blacklist"] is None
else set(cast(Iterable, config["blacklist"]))
)
match (w_list, b_list):
case (None, None):
return True
case (None, set()) if not b_list:
return True
case (set(), _) if not w_list:
return False
case (None, _):
return table_name not in b_list # type: ignore[operator]
case (_, None):
return table_name in w_list # type: ignore[operator]
case _:
return table_name in w_list - b_list # type: ignore[operator]
Py-linq-sql Exceptions
All exceptions for the project py_linq_sql.
ReturnEmptyEnumerable (Warning)
Raised when we get an empty enumerable in the record.
Source code in py_linq_sql/exception/exception.py
class ReturnEmptyEnumerable(Warning):
"""Raised when we get an empty enumerable in the record."""
PyLinqSQLError (Exception)
Base exception for the project.
Source code in py_linq_sql/exception/exception.py
class PyLinqSQLError(Exception):
"""Base exception for the project."""
PSQLConnectionError (PyLinqSQLError)
Error raised if we can't connect to a database.
Source code in py_linq_sql/exception/exception.py
class PSQLConnectionError(PyLinqSQLError):
"""Error raised if we can't connect to a database."""
ConfigPermissionError (PyLinqSQLError)
Error raised if you does not have permission to access to something.
Source code in py_linq_sql/exception/exception.py
class ConfigPermissionError(PyLinqSQLError):
"""Error raised if you does not have permission to access to something."""
TablePermissionDeniedError (ConfigPermissionError)
Raised when a table is not accessible according to the black and white lists.
Source code in py_linq_sql/exception/exception.py
class TablePermissionDeniedError(ConfigPermissionError):
"""Raised when a table is not accessible according to the black and white lists."""
def __init__(self, table: str) -> None:
"""Initialize a PermissionDeniedError."""
super().__init__(
f"Permission Denied for the table {table}. "
"Check your white and black lists.",
)
__init__(self, table)
special
Initialize a PermissionDeniedError.
Source code in py_linq_sql/exception/exception.py
def __init__(self, table: str) -> None:
"""Initialize a PermissionDeniedError."""
super().__init__(
f"Permission Denied for the table {table}. "
"Check your white and black lists.",
)
ReadOnlyPermissionDeniedError (ConfigPermissionError)
Raised when a table is not accessible according to the black and white lists.
Source code in py_linq_sql/exception/exception.py
class ReadOnlyPermissionDeniedError(ConfigPermissionError):
"""Raised when a table is not accessible according to the black and white lists."""
def __init__(self) -> None:
"""Initialize a PermissionDeniedError."""
super().__init__(
"Permission Denied write to the database. "
"Check your read only configuration.",
)
__init__(self)
special
Initialize a PermissionDeniedError.
Source code in py_linq_sql/exception/exception.py
def __init__(self) -> None:
"""Initialize a PermissionDeniedError."""
super().__init__(
"Permission Denied write to the database. "
"Check your read only configuration.",
)
InputError (PyLinqSQLError)
Error raised if input argument is invalid.
Source code in py_linq_sql/exception/exception.py
class InputError(PyLinqSQLError):
"""Error raised if input argument is invalid."""
ColumnNameError (InputError)
Raised when a custom column name is not in the good format.
Source code in py_linq_sql/exception/exception.py
class ColumnNameError(InputError):
"""Raised when a custom column name is not in the good format."""
def __init__(self, name: str) -> None:
"""Initialize a ColumnNameError."""
super().__init__(
f"The column name '{name}' isn't a valid column name. "
"A valid column name begin by a letter and is followed by "
"letter, number or '_'. The max size is 58 bytes. "
"The max size of a column name is 63 bytes for the database. "
"But here we limit it to 58 bytes "
"because we keep 1 bytes for a potential prefix of psql "
"and 4 bytes for a suffix if there are several identical column names.",
)
__init__(self, name)
special
Initialize a ColumnNameError.
Source code in py_linq_sql/exception/exception.py
def __init__(self, name: str) -> None:
"""Initialize a ColumnNameError."""
super().__init__(
f"The column name '{name}' isn't a valid column name. "
"A valid column name begin by a letter and is followed by "
"letter, number or '_'. The max size is 58 bytes. "
"The max size of a column name is 63 bytes for the database. "
"But here we limit it to 58 bytes "
"because we keep 1 bytes for a potential prefix of psql "
"and 4 bytes for a suffix if there are several identical column names.",
)
NegativeNumberError (ArithmeticError, InputError)
Error raised when we try to pass a negative number to a function.
If this function accepts only positive numbers.
Source code in py_linq_sql/exception/exception.py
class NegativeNumberError(ArithmeticError, InputError):
"""
Error raised when we try to pass a negative number to a function.
If this function accepts only positive numbers.
"""
def __init__(self) -> None:
"""Initialize a NegativeNumberError."""
super().__init__("numbers must be positive.")
__init__(self)
special
Initialize a NegativeNumberError.
Source code in py_linq_sql/exception/exception.py
def __init__(self) -> None:
"""Initialize a NegativeNumberError."""
super().__init__("numbers must be positive.")
TableError (InputError)
Raised when we try to reference a table not selected after a join.
Source code in py_linq_sql/exception/exception.py
class TableError(InputError):
"""Raised when we try to reference a table not selected after a join."""
def __init__(self, table: str, good_tables: list[str]) -> None:
"""Initialize a TableError."""
super().__init__(
"You try to make a command after a join. When you do that you must "
"give the table EXPLICITLY in the lambda.\n"
"Example:\n"
"\t.where(lambda x: x.satellite.data.name)\n"
f"Here you have referred to {table} while the tables to select in the join "
f"are {good_tables}.\n",
)
__init__(self, table, good_tables)
special
Initialize a TableError.
Source code in py_linq_sql/exception/exception.py
def __init__(self, table: str, good_tables: list[str]) -> None:
"""Initialize a TableError."""
super().__init__(
"You try to make a command after a join. When you do that you must "
"give the table EXPLICITLY in the lambda.\n"
"Example:\n"
"\t.where(lambda x: x.satellite.data.name)\n"
f"Here you have referred to {table} while the tables to select in the join "
f"are {good_tables}.\n",
)
TypeOperatorError (InputError, TypeError)
Error raised when we try to cast a value in a unknown type.
Source code in py_linq_sql/exception/exception.py
class TypeOperatorError(InputError, TypeError):
"""Error raised when we try to cast a value in a unknown type."""
def __init__(self, expected: list[type], actual: type) -> None:
"""Initialize a TypeOperatorError."""
super().__init__(
f"Wrong type, only :\n" f"{expected} " f"can be used here. Not {actual}.",
)
__init__(self, expected, actual)
special
Initialize a TypeOperatorError.
Source code in py_linq_sql/exception/exception.py
def __init__(self, expected: list[type], actual: type) -> None:
"""Initialize a TypeOperatorError."""
super().__init__(
f"Wrong type, only :\n" f"{expected} " f"can be used here. Not {actual}.",
)
PreExecutionError (PyLinqSQLError)
Error raised when we can't prepare the execution.
Source code in py_linq_sql/exception/exception.py
class PreExecutionError(PyLinqSQLError):
"""Error raised when we can't prepare the execution."""
ActionError (PreExecutionError)
Error raised when we try to execute a request without action.
Source code in py_linq_sql/exception/exception.py
class ActionError(PreExecutionError):
"""Error raised when we try to execute a request without action."""
def __init__(self) -> None:
"""Initialize an ActionError."""
super().__init__(
"You have to make an action command before execute an SQLEnumerable.",
)
__init__(self)
special
Initialize an ActionError.
Source code in py_linq_sql/exception/exception.py
def __init__(self) -> None:
"""Initialize an ActionError."""
super().__init__(
"You have to make an action command before execute an SQLEnumerable.",
)
EmptyInputError (PreExecutionError)
Error raised when we try to call a command with an empty argument.
Source code in py_linq_sql/exception/exception.py
class EmptyInputError(PreExecutionError):
"""Error raised when we try to call a command with an empty argument."""
def __init__(self, name_of_cmd: str) -> None:
"""Initialize an EmptyInputError."""
super().__init__(f"Input in {name_of_cmd} are empty.")
__init__(self, name_of_cmd)
special
Initialize an EmptyInputError.
Source code in py_linq_sql/exception/exception.py
def __init__(self, name_of_cmd: str) -> None:
"""Initialize an EmptyInputError."""
super().__init__(f"Input in {name_of_cmd} are empty.")
EmptyRecordError (PreExecutionError)
Error raised when the record of a request is empty.
Source code in py_linq_sql/exception/exception.py
class EmptyRecordError(PreExecutionError):
"""Error raised when the record of a request is empty."""
def __init__(self) -> None:
"""Initialize an EmptyRecordError."""
super().__init__("Record are None.")
__init__(self)
special
Initialize an EmptyRecordError.
Source code in py_linq_sql/exception/exception.py
def __init__(self) -> None:
"""Initialize an EmptyRecordError."""
super().__init__("Record are None.")
BuildError (PyLinqSQLError)
Error raised when something wrong when we build the request.
Source code in py_linq_sql/exception/exception.py
class BuildError(PyLinqSQLError):
"""Error raised when something wrong when we build the request."""
DeleteError (PyLinqSQLError)
Raised when we try to make a delete with a predicate and armageddon == True.
Source code in py_linq_sql/exception/exception.py
class DeleteError(PyLinqSQLError):
"""Raised when we try to make a delete with a predicate and armageddon == True."""
def __init__(self, other_cmd: str) -> None:
"""Initialize a DeleteError."""
super().__init__(
"You can't make a delete with "
f"a {other_cmd} and armageddon set at True. You need to choose.",
)
__init__(self, other_cmd)
special
Initialize a DeleteError.
Source code in py_linq_sql/exception/exception.py
def __init__(self, other_cmd: str) -> None:
"""Initialize a DeleteError."""
super().__init__(
"You can't make a delete with "
f"a {other_cmd} and armageddon set at True. You need to choose.",
)
LengthMismatchError (PyLinqSQLError)
Error raised when length of lambda are not equal.
Source code in py_linq_sql/exception/exception.py
class LengthMismatchError(PyLinqSQLError):
"""Error raised when length of lambda are not equal."""
def __init__(self, *args: str) -> None:
"""Initialize a LengthMismatchError."""
msg = "Size of "
for idx, _input in enumerate(args):
msg += _input
if not idx == len(args) - 1:
msg += ", "
msg += " are not equal"
super().__init__(msg)
__init__(self, *args)
special
Initialize a LengthMismatchError.
Source code in py_linq_sql/exception/exception.py
def __init__(self, *args: str) -> None:
"""Initialize a LengthMismatchError."""
msg = "Size of "
for idx, _input in enumerate(args):
msg += _input
if not idx == len(args) - 1:
msg += ", "
msg += " are not equal"
super().__init__(msg)
NeedWhereError (PyLinqSQLError)
Raised when we try to make a delete without armageddon and without where.
Source code in py_linq_sql/exception/exception.py
class NeedWhereError(PyLinqSQLError):
"""Raised when we try to make a delete without armageddon and without where."""
def __init__(self) -> None:
"""Initialize a NeedWhereError."""
super().__init__(
"You need to make a where before or after delete "
"or give a predicate to delete if you don't use "
"the parameter `armageddon` to say 'I want to delete all my table'.",
)
__init__(self)
special
Initialize a NeedWhereError.
Source code in py_linq_sql/exception/exception.py
def __init__(self) -> None:
"""Initialize a NeedWhereError."""
super().__init__(
"You need to make a where before or after delete "
"or give a predicate to delete if you don't use "
"the parameter `armageddon` to say 'I want to delete all my table'.",
)
TooManyReturnValueError (TypeError, PyLinqSQLError)
Error raised when we pass to many value to a MagicDotPaths.
Source code in py_linq_sql/exception/exception.py
class TooManyReturnValueError(TypeError, PyLinqSQLError):
"""Error raised when we pass to many value to a MagicDotPaths."""
def __init__(self, name: str) -> None:
"""Initialize a TooManyReturnValueError."""
super().__init__(f"{name} take only one lambda, only one modification.")
__init__(self, name)
special
Initialize a TooManyReturnValueError.
Source code in py_linq_sql/exception/exception.py
def __init__(self, name: str) -> None:
"""Initialize a TooManyReturnValueError."""
super().__init__(f"{name} take only one lambda, only one modification.")
ExecutionError (PyLinqSQLError)
Error raised if we can't execute a request.
Source code in py_linq_sql/exception/exception.py
class ExecutionError(PyLinqSQLError):
"""Error raised if we can't execute a request."""
def __init__(
self,
err: Exception,
command: str | None,
context: str,
executed: bool,
) -> None:
"""Initialize an ExecutionError."""
exec_ = "wasn't" if not executed else "was"
super().__init__(
f"The command {command} "
f"{exec_} executed.\n"
f"Context : {context}.\n"
"An exception has occurred: "
f"{err.__str__()}\n"
f"Exception TYPE: {type(err)}\n",
)
__init__(self, err, command, context, executed)
special
Initialize an ExecutionError.
Source code in py_linq_sql/exception/exception.py
def __init__(
self,
err: Exception,
command: str | None,
context: str,
executed: bool,
) -> None:
"""Initialize an ExecutionError."""
exec_ = "wasn't" if not executed else "was"
super().__init__(
f"The command {command} "
f"{exec_} executed.\n"
f"Context : {context}.\n"
"An exception has occurred: "
f"{err.__str__()}\n"
f"Exception TYPE: {type(err)}\n",
)
CursorCloseError (ExecutionError)
Error raised when we try to close a cursor already close..
Source code in py_linq_sql/exception/exception.py
class CursorCloseError(ExecutionError):
"""Error raised when we try to close a cursor already close.."""
DatabError (ExecutionError)
Error raised when something wrong with the database.
Source code in py_linq_sql/exception/exception.py
class DatabError(ExecutionError):
"""Error raised when something wrong with the database."""
EmptyQueryError (ExecutionError)
Error raised when we try to execute a empty query.
Source code in py_linq_sql/exception/exception.py
class EmptyQueryError(ExecutionError):
"""Error raised when we try to execute a empty query."""
FetchError (ExecutionError)
Error raised when we can't fetch the record of an execution.
Source code in py_linq_sql/exception/exception.py
class FetchError(ExecutionError):
"""Error raised when we can't fetch the record of an execution."""
LegalityError (PyLinqSQLError)
Error raised when we does not respect the legality rules.
Source code in py_linq_sql/exception/exception.py
class LegalityError(PyLinqSQLError):
"""Error raised when we does not respect the legality rules."""
AlreadyExecutedError (LegalityError)
Raised when we try use an SQLEnumerable who has been already executed.
Source code in py_linq_sql/exception/exception.py
class AlreadyExecutedError(LegalityError):
"""Raised when we try use an SQLEnumerable who has been already executed."""
def __init__(self) -> None:
"""Initialize an AlreadyExecutedError."""
super().__init__(
"You can't re use an SQL_Enumerable who has been already executed.",
)
__init__(self)
special
Initialize an AlreadyExecutedError.
Source code in py_linq_sql/exception/exception.py
def __init__(self) -> None:
"""Initialize an AlreadyExecutedError."""
super().__init__(
"You can't re use an SQL_Enumerable who has been already executed.",
)
AlterError (LegalityError)
Error raised when we try to make a action command after a alter command.
Source code in py_linq_sql/exception/exception.py
class AlterError(LegalityError):
"""Error raised when we try to make a action command after a alter command."""
def __init__(self, self_name: str) -> None:
"""Initialize an AlterError."""
super().__init__(f"You can't make {self_name} after an alter command.")
__init__(self, self_name)
special
Initialize an AlterError.
Source code in py_linq_sql/exception/exception.py
def __init__(self, self_name: str) -> None:
"""Initialize an AlterError."""
super().__init__(f"You can't make {self_name} after an alter command.")
GroupByWithJoinError (LegalityError)
Raised when we try to make a group by after a join.
Source code in py_linq_sql/exception/exception.py
class GroupByWithJoinError(LegalityError):
"""Raised when we try to make a group by after a join."""
def __init__(self) -> None:
"""Initialize a GroupByWithJoinError."""
super().__init__("You can't make a group_by after a join, use group_join().")
__init__(self)
special
Initialize a GroupByWithJoinError.
Source code in py_linq_sql/exception/exception.py
def __init__(self) -> None:
"""Initialize a GroupByWithJoinError."""
super().__init__("You can't make a group_by after a join, use group_join().")
MoreThanZeroError (LegalityError)
Error raised when we make command who must be alone and this isn't the case.
Source code in py_linq_sql/exception/exception.py
class MoreThanZeroError(LegalityError):
"""Error raised when we make command who must be alone and this isn't the case."""
def __init__(self, name_of_cmd: str) -> None:
"""Initialize a MoreThanZeroError."""
super().__init__(f"You can't make {name_of_cmd} command after other command.")
__init__(self, name_of_cmd)
special
Initialize a MoreThanZeroError.
Source code in py_linq_sql/exception/exception.py
def __init__(self, name_of_cmd: str) -> None:
"""Initialize a MoreThanZeroError."""
super().__init__(f"You can't make {name_of_cmd} command after other command.")
NeedSelectError (LegalityError)
Raised when we try to make a command who need a select and we don't have it.
Source code in py_linq_sql/exception/exception.py
class NeedSelectError(LegalityError):
"""Raised when we try to make a command who need a select and we don't have it."""
def __init__(self, cmd_name: str) -> None:
"""Initialize a NeedSelectError."""
super().__init__(f"All SQLEnumerable in {cmd_name} must be a SELECT request.")
__init__(self, cmd_name)
special
Initialize a NeedSelectError.
Source code in py_linq_sql/exception/exception.py
def __init__(self, cmd_name: str) -> None:
"""Initialize a NeedSelectError."""
super().__init__(f"All SQLEnumerable in {cmd_name} must be a SELECT request.")
NoMaxOrMinAfterLimitOffsetError (LegalityError)
Error when we try to make a max or a min after a limit offset command.
In SQL MAX and MIN return only one element it's useless to make take or skip after. If we want take only 4 elements in the table and get the max we can make the max (or min) after the execution with Enumerable.max().
If we our take()
takes a very large number and we want the max
to be executed by the server we must execute an explicit nested fquery.
Examples:
With a not too big take()
:
>>> record = SQLEnumerable(con, table).take(183).execute()
>>> record.max(lambda x: x["mass"])
With a large take()
:
>>> sqle_1 = SQLEnumerable(con, table).take(4691)
>>> record = (
... SQLEnumerable(con, sqle_1)
... .max(lambda x: x.mass)
... .execute()
... )
Source code in py_linq_sql/exception/exception.py
class NoMaxOrMinAfterLimitOffsetError(LegalityError):
"""
Error when we try to make a max or a min after a limit offset command.
In SQL MAX and MIN return only one element it's useless to make take or skip after.
If we want take only 4 elements in the table and get the max we can make the max
(or min) after the execution with Enumerable.max().
If we our `take()` takes a very large number and we want the max
to be executed by the server we must execute an explicit nested fquery.
Examples:
With a not too big `take()`:
>>> record = SQLEnumerable(con, table).take(183).execute() # doctest: +SKIP
>>> record.max(lambda x: x["mass"]) # doctest: +SKIP
With a large `take()`:
>>> sqle_1 = SQLEnumerable(con, table).take(4691) # doctest: +SKIP
>>> record = ( # doctest: +SKIP
... SQLEnumerable(con, sqle_1) # doctest: +SKIP
... .max(lambda x: x.mass) # doctest: +SKIP
... .execute() # doctest: +SKIP
... ) # doctest: +SKIP
"""
def __init__(self, cmd_name: str) -> None:
"""Initialize a NoMaxOrMinAfterLimitOffsetError."""
super().__init__(f"You can't make {cmd_name} after a limit or offset.")
__init__(self, cmd_name)
special
Initialize a NoMaxOrMinAfterLimitOffsetError.
Source code in py_linq_sql/exception/exception.py
def __init__(self, cmd_name: str) -> None:
"""Initialize a NoMaxOrMinAfterLimitOffsetError."""
super().__init__(f"You can't make {cmd_name} after a limit or offset.")
OneError (LegalityError)
Error raised when we try to make a command after a one command.
Source code in py_linq_sql/exception/exception.py
class OneError(LegalityError):
"""Error raised when we try to make a command after a one command."""
def __init__(self) -> None:
"""Initialize a OneError."""
super().__init__("You can't make command after one commands.")
__init__(self)
special
Initialize a OneError.
Source code in py_linq_sql/exception/exception.py
def __init__(self) -> None:
"""Initialize a OneError."""
super().__init__("You can't make command after one commands.")
OtherThanWhereError (LegalityError)
Error raised when we get commands other than where.
For functions who only accepts where commands.
Source code in py_linq_sql/exception/exception.py
class OtherThanWhereError(LegalityError):
"""
Error raised when we get commands other than where.
For functions who only accepts where commands.
"""
def __init__(self) -> None:
"""Initialize an OtherThanWhereError."""
super().__init__("Other than where commands in list of commands.")
__init__(self)
special
Initialize an OtherThanWhereError.
Source code in py_linq_sql/exception/exception.py
def __init__(self) -> None:
"""Initialize an OtherThanWhereError."""
super().__init__("Other than where commands in list of commands.")
SelectError (LegalityError)
Error raised when we try to make a action command after a select command.
Source code in py_linq_sql/exception/exception.py
class SelectError(LegalityError):
"""Error raised when we try to make a action command after a select command."""
def __init__(self, self_name: str) -> None:
"""Initialize a SelectError."""
super().__init__(f"You can't make {self_name} after select.")
__init__(self, self_name)
special
Initialize a SelectError.
Source code in py_linq_sql/exception/exception.py
def __init__(self, self_name: str) -> None:
"""Initialize a SelectError."""
super().__init__(f"You can't make {self_name} after select.")
TerminalError (LegalityError)
Error raised when we try to make a command after a terminal command.
Source code in py_linq_sql/exception/exception.py
class TerminalError(LegalityError):
"""Error raised when we try to make a command after a terminal command."""
def __init__(self, name_of_cmd: str) -> None:
"""Initialize a TerminalError."""
super().__init__(
f"You can't make {name_of_cmd} command after a terminal command.",
)
__init__(self, name_of_cmd)
special
Initialize a TerminalError.
Source code in py_linq_sql/exception/exception.py
def __init__(self, name_of_cmd: str) -> None:
"""Initialize a TerminalError."""
super().__init__(
f"You can't make {name_of_cmd} command after a terminal command.",
)
UnknownCommandTypeError (TypeError, PyLinqSQLError)
Error raised when we try to build an unknown command.
Source code in py_linq_sql/exception/exception.py
class UnknownCommandTypeError(TypeError, PyLinqSQLError):
"""Error raised when we try to build an unknown command."""
# The following case is just an other security layers,
# but we can't go in this case for the moment.
def __init__(self, cmd_type: str) -> None: # pragma: no cover
"""Initialize an UnknownError."""
super().__init__(f"Unknown command type : {cmd_type}")
__init__(self, cmd_type)
special
Initialize an UnknownError.
Source code in py_linq_sql/exception/exception.py
def __init__(self, cmd_type: str) -> None: # pragma: no cover
"""Initialize an UnknownError."""
super().__init__(f"Unknown command type : {cmd_type}")
Structure of Exceptions
.
├── Warning
| └── ReturnEmptyEnumerable
├── PyLinqSQLError
| ├── PSQLConnectionError
| ├── ConfigPermissionError
| | ├── TablePermissionDeniedError
| | └── ReadOnlyPermissionDeniedError
| ├── InputError
| | ├── ColumnNameError
| | ├── NegativeNumberError
| | ├── TableError
| | └── TypeOperatorError
| ├── PreExecutionError
| | ├── ActionError
| | ├── EmptyInputError
| | └── EmptyRecordError
| ├── BuildError
| | ├── DeleteError
| | ├── LenghtMismatchError
| | ├── NeedWhereError
| | └── TooManyReturnValueError
| ├── ExecutionError
| | ├── CursorCloseError
| | ├── EmptyQueryError
| | ├── DatabError
| | └── FetchError
| ├── LegalityError
| | ├── AlreadyExecutedError
| | ├── AlterError
| | ├── GroupByWithJoinError
| | ├── MoreThanZeroError
| | ├── NeedSelectError
| | ├── NoMaxOrMinAfterLimitOffsetError
| | ├── OneError
| | ├── OtherThanWhereError
| | ├── SelectError
| | └── TerminalError
| └── UnknownCommandTypeError