where

Where clauses for filtering queries.

And

Bases: ClauseBase

AND two clauses.

Source code in src/embar/query/where.py

class And(ClauseBase):
    """
    AND two clauses.
    """

    left: ClauseBase
    right: ClauseBase

    def __init__(self, left: ClauseBase, right: ClauseBase):
        self.left = left
        self.right = right

    @override
    def sql(self, get_count: GetCount) -> QuerySingle:
        left = self.left.sql(get_count)
        right = self.right.sql(get_count)
        params = {**left.params, **right.params}
        sql = f"{left.sql} AND {right.sql}"
        return QuerySingle(sql=sql, params=params)

Between

Bases: ClauseBase

Checks if a column value is between two values (inclusive).

Source code in src/embar/query/where.py

class Between[T: PyType](ClauseBase):
    """
    Checks if a column value is between two values (inclusive).
    """

    column: ColumnInfo
    lower: PyType
    upper: PyType

    def __init__(self, column: Column[T], lower: T, upper: T):
        self.column = column.info
        self.lower = lower
        self.upper = upper

    @override
    def sql(self, get_count: GetCount) -> QuerySingle:
        count = get_count()
        lower_name = f"between_lower_{self.column.name}_{count}"
        upper_name = f"between_upper_{self.column.name}_{count}"
        return QuerySingle(
            sql=f"{self.column.fqn()} BETWEEN %({lower_name})s AND %({upper_name})s",
            params={lower_name: self.lower, upper_name: self.upper},
        )

Eq

Bases: ClauseBase

Checks if a column value is equal to another column or a passed param.

Right now the left must always be a column, maybe that must be loosened.

Source code in src/embar/query/where.py

class Eq(ClauseBase):
    """
    Checks if a column value is equal to another column or a passed param.

    Right now the left must always be a column, maybe that must be loosened.
    """

    left: ColumnInfo | ClauseBase
    right: ColumnInfo | PyType

    @overload
    def __init__[T: PyType](self, left: Column[T], right: Column[T]) -> None: ...
    @overload
    def __init__[T: PyType](self, left: Column[T], right: T) -> None: ...
    @overload
    def __init__(self, left: ClauseBase, right: ColumnBase | PyType) -> None: ...
    def __init__(self, left: Column[Any] | ClauseBase, right: Column[Any] | PyType) -> None:
        self.left = left.info if isinstance(left, Column) else left
        self.right = right.info if isinstance(right, Column) else right

    @override
    def sql(self, get_count: GetCount) -> QuerySingle:
        return _gen_comparison_sql(self.left, self.right, "=", "eq", get_count)

Exists

Bases: ClauseBase

Check if a subquery result exists.

Source code in src/embar/query/where.py

class Exists(ClauseBase):
    """
    Check if a subquery result exists.
    """

    query: SqlAble

    def __init__(self, query: SqlAble):
        self.query = query

    @override
    def sql(self, get_count: GetCount) -> QuerySingle:
        query = self.query.sql()
        return QuerySingle(f"EXISTS ({query.sql})", query.params)

Gt

Bases: ClauseBase

Checks if a column value is greater than another column or a passed param.

Source code in src/embar/query/where.py

class Gt(ClauseBase):
    """
    Checks if a column value is greater than another column or a passed param.
    """

    left: ColumnInfo | ClauseBase
    right: ColumnInfo | PyType

    @overload
    def __init__[T: PyType](self, left: Column[T], right: Column[T]) -> None: ...
    @overload
    def __init__[T: PyType](self, left: Column[T], right: T) -> None: ...
    @overload
    def __init__(self, left: ClauseBase, right: ColumnBase | PyType) -> None: ...
    def __init__(self, left: Column[Any] | ClauseBase, right: Column[Any] | PyType) -> None:
        self.left = left.info if isinstance(left, Column) else left
        self.right = right.info if isinstance(right, Column) else right

    @override
    def sql(self, get_count: GetCount) -> QuerySingle:
        return _gen_comparison_sql(self.left, self.right, ">", "gt", get_count)

Gte

Bases: ClauseBase

Checks if a column value is greater than or equal to another column or a passed param.

Source code in src/embar/query/where.py

class Gte(ClauseBase):
    """
    Checks if a column value is greater than or equal to another column or a passed param.
    """

    left: ColumnInfo | ClauseBase
    right: ColumnInfo | PyType

    @overload
    def __init__[T: PyType](self, left: Column[T], right: Column[T]) -> None: ...
    @overload
    def __init__[T: PyType](self, left: Column[T], right: T) -> None: ...
    @overload
    def __init__(self, left: ClauseBase, right: ColumnBase | PyType) -> None: ...
    def __init__(self, left: Column[Any] | ClauseBase, right: Column[Any] | PyType) -> None:
        self.left = left.info if isinstance(left, Column) else left
        self.right = right.info if isinstance(right, Column) else right

    @override
    def sql(self, get_count: GetCount) -> QuerySingle:
        return _gen_comparison_sql(self.left, self.right, ">=", "gte", get_count)

Ilike

Bases: ClauseBase

Case-insensitive LIKE pattern matching.

Source code in src/embar/query/where.py

class Ilike[T: PyType](ClauseBase):
    """
    Case-insensitive LIKE pattern matching.
    """

    left: ColumnInfo
    right: T | ColumnInfo

    def __init__(self, left: Column[T], right: T | Column[T]):
        self.left = left.info
        self.right = right.info if isinstance(right, Column) else right

    @override
    def sql(self, get_count: GetCount) -> QuerySingle:
        count = get_count()
        name = f"ilike_{self.left.name}_{count}"
        if isinstance(self.right, ColumnInfo):
            return QuerySingle(sql=f"{self.left.fqn()} ILIKE {self.right.fqn()}")

        return QuerySingle(sql=f"{self.left.fqn()} ILIKE %({name})s", params={name: self.right})

InArray

Bases: ClauseBase

Checks if a column value is in a list of values.

Source code in src/embar/query/where.py

class InArray[T: PyType](ClauseBase):
    """
    Checks if a column value is in a list of values.
    """

    column: ColumnInfo
    values: list[T]

    def __init__(self, column: Column[T], values: list[T]):
        self.column = column.info
        self.values = values

    @override
    def sql(self, get_count: GetCount) -> QuerySingle:
        count = get_count()
        name = f"in_{self.column.name}_{count}"
        return QuerySingle(sql=f"{self.column.fqn()} = ANY(%({name})s)", params={name: self.values})

IsNotNull

Bases: ClauseBase

Checks if a column value is NOT NULL.

Source code in src/embar/query/where.py

class IsNotNull(ClauseBase):
    """
    Checks if a column value is NOT NULL.
    """

    column: ColumnInfo

    def __init__(self, column: Column[Any]):
        self.column = column.info

    @override
    def sql(self, get_count: GetCount) -> QuerySingle:
        return QuerySingle(sql=f"{self.column.fqn()} IS NOT NULL")

IsNull

Bases: ClauseBase

Checks if a column value is NULL.

Source code in src/embar/query/where.py

class IsNull(ClauseBase):
    """
    Checks if a column value is NULL.
    """

    column: ColumnInfo

    def __init__(self, column: Column[Any]):
        self.column = column.info

    @override
    def sql(self, get_count: GetCount) -> QuerySingle:
        return QuerySingle(sql=f"{self.column.fqn()} IS NULL")

Lt

Bases: ClauseBase

Checks if a column value is less than another column or a passed param.

Source code in src/embar/query/where.py

class Lt(ClauseBase):
    """
    Checks if a column value is less than another column or a passed param.
    """

    left: ColumnInfo | ClauseBase
    right: ColumnInfo | PyType

    @overload
    def __init__[T: PyType](self, left: Column[T], right: Column[T]) -> None: ...
    @overload
    def __init__[T: PyType](self, left: Column[T], right: T) -> None: ...
    @overload
    def __init__(self, left: ClauseBase, right: ColumnBase | PyType) -> None: ...
    def __init__(self, left: Column[Any] | ClauseBase, right: Column[Any] | PyType) -> None:
        self.left = left.info if isinstance(left, Column) else left
        self.right = right.info if isinstance(right, Column) else right

    @override
    def sql(self, get_count: GetCount) -> QuerySingle:
        return _gen_comparison_sql(self.left, self.right, "<", "lt", get_count)

Lte

Bases: ClauseBase

Checks if a column value is less than or equal to another column or a passed param.

Source code in src/embar/query/where.py

class Lte(ClauseBase):
    """
    Checks if a column value is less than or equal to another column or a passed param.
    """

    left: ColumnInfo | ClauseBase
    right: ColumnInfo | PyType

    @overload
    def __init__[T: PyType](self, left: Column[T], right: Column[T]) -> None: ...
    @overload
    def __init__[T: PyType](self, left: Column[T], right: T) -> None: ...
    @overload
    def __init__(self, left: ClauseBase, right: ColumnBase | PyType) -> None: ...
    def __init__(self, left: Column[Any] | ClauseBase, right: Column[Any] | PyType) -> None:
        self.left = left.info if isinstance(left, Column) else left
        self.right = right.info if isinstance(right, Column) else right

    @override
    def sql(self, get_count: GetCount) -> QuerySingle:
        return _gen_comparison_sql(self.left, self.right, "<=", "lte", get_count)

Ne

Bases: ClauseBase

Checks if a column value is not equal to another column or a passed param.

Source code in src/embar/query/where.py

class Ne(ClauseBase):
    """
    Checks if a column value is not equal to another column or a passed param.
    """

    left: ColumnInfo | ClauseBase
    right: ColumnInfo | PyType

    @overload
    def __init__[T: PyType](self, left: Column[T], right: Column[T]) -> None: ...
    @overload
    def __init__[T: PyType](self, left: Column[T], right: T) -> None: ...
    @overload
    def __init__(self, left: ClauseBase, right: ColumnBase | PyType) -> None: ...
    def __init__(self, left: Column[Any] | ClauseBase, right: Column[Any] | PyType) -> None:
        self.left = left.info if isinstance(left, Column) else left
        self.right = right.info if isinstance(right, Column) else right

    @override
    def sql(self, get_count: GetCount) -> QuerySingle:
        return _gen_comparison_sql(self.left, self.right, "!=", "ne", get_count)

Not

Bases: ClauseBase

Negates a where clause.

Source code in src/embar/query/where.py

class Not(ClauseBase):
    """
    Negates a where clause.
    """

    clause: ClauseBase

    def __init__(self, clause: ClauseBase):
        self.clause = clause

    @override
    def sql(self, get_count: GetCount) -> QuerySingle:
        inner = self.clause.sql(get_count)
        return QuerySingle(sql=f"NOT ({inner.sql})", params=inner.params)

NotBetween

Bases: ClauseBase

Checks if a column value is not between two values (inclusive).

Source code in src/embar/query/where.py

class NotBetween[T: PyType](ClauseBase):
    """
    Checks if a column value is not between two values (inclusive).
    """

    column: ColumnInfo
    lower: PyType
    upper: PyType

    def __init__(self, column: Column[T], lower: T, upper: T):
        self.column = column.info
        self.lower = lower
        self.upper = upper

    @override
    def sql(self, get_count: GetCount) -> QuerySingle:
        count = get_count()
        lower_name = f"notbetween_lower_{self.column.name}_{count}"
        upper_name = f"notbetween_upper_{self.column.name}_{count}"
        return QuerySingle(
            sql=f"{self.column.fqn()} NOT BETWEEN %({lower_name})s AND %({upper_name})s",
            params={lower_name: self.lower, upper_name: self.upper},
        )

NotExists

Bases: ClauseBase

Check if a subquery result does not exist.

Source code in src/embar/query/where.py

class NotExists(ClauseBase):
    """
    Check if a subquery result does not exist.
    """

    query: SqlAble

    def __init__(self, query: SqlAble):
        self.query = query

    @override
    def sql(self, get_count: GetCount) -> QuerySingle:
        query = self.query.sql()
        return QuerySingle(f"NOT EXISTS ({query.sql})", query.params)

NotInArray

Bases: ClauseBase

Checks if a column value is not in a list of values.

Source code in src/embar/query/where.py

class NotInArray[T: PyType](ClauseBase):
    """
    Checks if a column value is not in a list of values.
    """

    column: ColumnInfo
    values: list[T]

    def __init__(self, column: Column[T], values: list[T]):
        self.column = column.info
        self.values = values

    @override
    def sql(self, get_count: GetCount) -> QuerySingle:
        count = get_count()
        name = f"notin_{self.column.name}_{count}"
        return QuerySingle(sql=f"{self.column.fqn()} != ALL(%({name})s)", params={name: self.values})

NotLike

Bases: ClauseBase

Negated LIKE pattern matching.

Source code in src/embar/query/where.py

class NotLike[T: PyType](ClauseBase):
    """
    Negated LIKE pattern matching.
    """

    left: ColumnInfo
    right: T | ColumnInfo

    def __init__(self, left: Column[T], right: T | Column[T]):
        self.left = left.info
        self.right = right.info if isinstance(right, Column) else right

    @override
    def sql(self, get_count: GetCount) -> QuerySingle:
        count = get_count()
        name = f"notlike_{self.left.name}_{count}"
        if isinstance(self.right, ColumnInfo):
            return QuerySingle(sql=f"{self.left.fqn()} NOT LIKE {self.right.fqn()}")

        return QuerySingle(sql=f"{self.left.fqn()} NOT LIKE %({name})s", params={name: self.right})

Or

Bases: ClauseBase

OR two clauses.

Source code in src/embar/query/where.py

class Or(ClauseBase):
    """
    OR two clauses.
    """

    left: ClauseBase
    right: ClauseBase

    def __init__(self, left: ClauseBase, right: ClauseBase):
        self.left = left
        self.right = right

    @override
    def sql(self, get_count: GetCount) -> QuerySingle:
        left = self.left.sql(get_count)
        right = self.right.sql(get_count)
        params = {**left.params, **right.params}
        sql = f"{left.sql} OR {right.sql}"
        return QuerySingle(sql=sql, params=params)