Base Selector

This module provides a base selector for SQLAlchemy

class loglan_core.addons.base_selector.BaseSelector(model, is_sqlite=False, case_sensitive=False, disable_model_check=False)[source]

Bases: object

A custom base selector that inherits from SQLAlchemy’s Select class. This class provides methods to execute a session and fetch results in different ways. It also provides a way to fetch many results.

Methods:

execute(session: Session) -> ResultProxy:: Executes the session and returns the result.
all(session: Session) -> List[ResultRow]:: Executes the session and returns all the results.
scalar(session: Session) -> Any:: Executes the session and returns a scalar result.
fetchmany(session: Session, size: int | None = None) -> List[ResultRow]:: Executes the session and fetches a specified number of results.

__init__(model, is_sqlite=False, case_sensitive=False, disable_model_check=False)[source]

Initializes the Selector.

Parameters:

model (Type) – The SQLAlchemy model class to select from.
is_sqlite (bool) – Flag indicating if the database is SQLite.
case_sensitive (bool) – Flag indicating if the queries should be case-sensitive.
disable_model_check (bool) – Flag indicating if the model check should be disabled.

_get_column(key)[source]

Get the column from the model.

Parameters:: key (str | InstrumentedAttribute) – The key of the column to filter by.
Raises:: AttributeError – If the model has no attribute with the given key.
Returns:: The SQLAlchemy column to filter by.
Return type:: InstrumentedAttribute

static _is_model_accepted(model, parent=<class 'loglan_core.base.BaseModel'>)[source]

Checks if the model is an instance of BaseModel or its child.

Raises:: ValueError – If the model is not an instance of BaseModel or its child.

all(session, unique=False)[source]

Executes the given session and returns all the results as a list.

Parameters:

session (Session) – SQLAlchemy Session object.
unique (bool, optional) – Flag indicating if the result should contain unique items.
False. (Defaults to)

Returns:

All the results of the executed session.

Return type:

List[ResultRow]

async all_async(session, unique=False)[source]

Executes the given session and returns all the results as a list.

Parameters:

session (AsyncSession) – SQLAlchemy Session object.
unique (bool, optional) – Flag indicating if the result should contain unique items.
False. (Defaults to)

Returns:

All the results of the executed session.

Return type:

List[ResultRow]

execute(session, unique=False)[source]

Executes the given session and returns the result.

Parameters:

session (Session) – SQLAlchemy Session object.
unique (bool, optional) – Flag indicating if the result should be unique.
False. (Defaults to)

Returns:

The result of the executed session.

Return type:

ResultProxy

async execute_async(session, unique=False)[source]

Executes the given session and returns the result.

Parameters:

session (AsyncSession) – SQLAlchemy Session object.
unique (bool, optional) – Flag indicating if the result should be unique.
False. (Defaults to)

Returns:

The result of the executed session.

Return type:

ResultProxy

fetchmany(session, size=None, unique=False)[source]

Executes the given session and fetches a specified number of results.

Parameters:

session (Session) – SQLAlchemy Session object.
size (int, optional) – Number of results to fetch. If None, fetches all results.
unique (bool, optional) – Flag indicating if the result should contain unique items.
False. (Defaults to)

Returns:

The fetched results.

Return type:

List[ResultRow]

async fetchmany_async(session, size=None, unique=False)[source]

Executes the given session and fetches a specified number of results.

Parameters:

session (AsyncSession) – SQLAlchemy Session object.
size (int, optional) – Number of results to fetch. If None, fetches all results.
unique (bool, optional) – Flag indicating if the result should contain unique items.
False. (Defaults to)

Returns:

The fetched results.

Return type:

List[ResultRow]

filter(*args)[source]

Filter results based on arbitrary keyword arguments.

Parameters:: *args – Column-value pairs to filter by.
Returns:: The current instance for method chaining.
Return type:: Self

filter_by(**kwargs)[source]

Filter results based on arbitrary keyword arguments.

Parameters:: *kwargs – Column-value pairs to filter by.
Returns:: The current instance for method chaining.
Return type:: Self

get_like_condition(key, value)[source]

Generate the condition based on settings provided by Selector instance like (is_sqlite, case_sensitive).

Parameters:

key (str | InstrumentedAttribute) – The key of the column to filter by.
value (Any) – The value to filter by.

Returns:

The SQLAlchemy condition to filter by.

Return type:

Condition

get_statement()[source]

Get the current SQLAlchemy _statement.

Returns:: The current SQLAlchemy _statement.
Return type:: Select

limit(limit)[source]

Limit the number of results returned.

Parameters:: limit (int) – The maximum number of results to return.
Returns:: The current instance for method chaining.
Return type:: Self

offset(offset)[source]

Set the offset for the results returned.

Parameters:: offset (int) – The number of results to skip before starting to return results.
Returns:: The current instance for method chaining.
Return type:: Self

order_by(*columns)[source]

Specify the order in which results should be returned.

Parameters:: *columns – The columns to order by.
Returns:: The current instance for method chaining.
Return type:: Self

scalar(session)[source]

Executes the given session and returns a scalar result.

Parameters:: session (Session) – SQLAlchemy Session object.
Returns:: The scalar result of the executed session.
Return type:: Any

async scalar_async(session)[source]

Executes the given session and returns a scalar result.

Parameters:: session (AsyncSession) – SQLAlchemy Session object.
Returns:: The scalar result of the executed session.
Return type:: Any

select_columns(*columns)[source]

Specify which columns to select without resetting the filters.

Parameters:: *columns (type[BaseModel]) – The columns to select.
Returns:: The current instance for method chaining.
Return type:: Self

where(*args)[source]

Filter results based on arbitrary keyword arguments.

Parameters:: *args – Column-value pairs to filter by.
Returns:: The current instance for method chaining.
Return type:: Self

where_like(**kwargs)[source]

Filter results based on arbitrary keyword arguments.: Use internal method _generate_column_condition to generate the condition based on settings provided by Selector instance like (is_sqlite, case_sensitive).

Parameters:: **kwargs – Column-value pairs to filter by.
Returns:: The current instance for method chaining.
Return type:: Self

with_relationships(selected=None)[source]

Adds relationships to the query.

Parameters:

selected (set[str]) – A set of relationship names to include.
included. (Defaults to None if all relationships should be)

Returns:

A query with the relationships added.

Return type:

Self