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.

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)
-- ... also fine for SQLite, giving 1 or None (no rows)

Others, including MySQL

SELECT EXISTS (SELECT 1 FROM table WHERE ...)
-- ... giving 1 or 0
-- ... fine for MySQL, but invalid syntax for SQL Server
-- ... also fine for SQLite, giving 1 or 0
cardinal_pythonlib.sqlalchemy.orm_query.exists_orm(session: Session, ormclass: Type[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: Type[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[List[Tuple[Any, ...]], List[str]][source]

Superseded. It used to be fine to use a Query object to run a SELECT statement. But as of SQLAlchemy 2.0 (or 1.4 with future=True), this has been removed.

Also, it isn’t worth coercing here. Some details are in the source code, but usually we are not seeking to run a query that fetches ORM objects themselves and then map those to fieldnames/values. Instead, we used to use a Query object made from selectable elements like columns and COUNT() clauses. That is what the select() system is meant for. So this code will now raise an error.