cardinal_pythonlib.sqlalchemy.orm_query


Original code copyright (C) 2009-2022 Rudolf Cardinal (rudolf@pobox.com).

This file is part of cardinal_pythonlib.

Licensed under the Apache License, Version 2.0 (the “License”); you may not use this file except in compliance with the License. You may obtain a copy of the License at

Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an “AS IS” BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions and limitations under the License.


Functions to perform and manipulate SQLAlchemy ORM queries.

class cardinal_pythonlib.sqlalchemy.orm_query.CountStarSpecializedQuery(*args, **kwargs)[source]

Optimizes COUNT(*) queries.

See https://stackoverflow.com/questions/12941416/how-to-count-rows-with-select-count-with-sqlalchemy

Example use:

q = CountStarSpecializedQuery([cls], session=dbsession)                .filter(cls.username == username)
return q.count_star()
count_star() int[source]

Implements the COUNT(*) specialization.

cardinal_pythonlib.sqlalchemy.orm_query.bool_from_exists_clause(session: Session, exists_clause: Exists) bool[source]

Database dialects are not consistent in how EXISTS clauses can be converted to a boolean answer. This function manages the inconsistencies.

See:

Specifically, we want this:

SQL Server

SELECT 1 WHERE EXISTS (SELECT 1 FROM table WHERE ...)
-- ... giving 1 or None (no rows)
-- ... fine for SQL Server, but invalid for MySQL (no FROM clause)

Others, including MySQL

SELECT EXISTS (SELECT 1 FROM table WHERE ...)
-- ... giving 1 or 0
-- ... fine for MySQL, but invalid syntax for SQL Server
cardinal_pythonlib.sqlalchemy.orm_query.exists_orm(session: Session, ormclass: DeclarativeMeta, *criteria: Any) bool[source]

Detects whether a database record exists for the specified ormclass and criteria.

Example usage:

bool_exists = exists_orm(session, MyClass, MyClass.myfield == value)
cardinal_pythonlib.sqlalchemy.orm_query.get_or_create(session: Session, model: DeclarativeMeta, defaults: Dict[str, Any] | None = None, **kwargs: Any) Tuple[Any, bool][source]

Fetches an ORM object from the database, or creates one if none existed.

Parameters:
  • session – an SQLAlchemy Session

  • model – an SQLAlchemy ORM class

  • defaults – default initialization arguments (in addition to relevant filter criteria) if we have to create a new instance

  • kwargs – optional filter criteria

Returns:

a tuple (instance, newly_created)

See https://stackoverflow.com/questions/2546207 (this function is a composite of several suggestions).

cardinal_pythonlib.sqlalchemy.orm_query.get_rows_fieldnames_from_query(session: Session | Engine | Connection, query: Query) Tuple[Sequence[Sequence[Any]], Sequence[str]][source]

Returns results and column names from a query.

Parameters:
  • session – SQLAlchemy Session, Engine, or Connection object

  • query – SQLAlchemy Query

Returns:

(rows, fieldnames) where rows is the usual set of results and fieldnames are the name of the result columns/fields.