cardinal_pythonlib.sqlalchemy.core_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.

Query helper functions using the SQLAlchemy Core.

Example of result types in SQLAlchemy 1.4+ and higher:

from typing import List
from sqlalchemy.engine.cursor import CursorResult
from sqlalchemy.engine.result import MappingResult, Result
from sqlalchemy.engine.row import Row, RowMapping

query = (
    select(text("*"))
    .select_from(table(some_tablename))
)

# As tuples:
result_1: CursorResult = session.execute(query)
# ... or, more generically, of type Result
like_unnamed_tuples: List[Row] = result_1.fetchall()

# Or:
result_2: Result = session.execute(query)
mapping_result: Mapping_Result = result_2.mappings()
like_dicts: List[RowMapping] = list(mapping_result)  # implicit fetchall()
# ... or could have done: like_dicts = result_2.mappings().fetchall()
cardinal_pythonlib.sqlalchemy.core_query.count_star(session: Session | Engine | Connection, tablename: str, *criteria: Any) → int[source]

Returns the result of COUNT(*) from the specified table (with additional WHERE criteria if desired).

Parameters:

  • session – SQLAlchemy Session, Engine (SQL Alchemy 1.4 only), or Connection object

  • tablename – name of the table

  • criteria – optional SQLAlchemy “where” criteria

Returns:

a scalar

cardinal_pythonlib.sqlalchemy.core_query.count_star_and_max(session: Session | Engine | Connection, tablename: str, maxfield: str, *criteria: Any) → Tuple[int, int | None][source]
Parameters:

  • session – SQLAlchemy Session, Engine (SQL Alchemy 1.4 only), or Connection object

  • tablename – name of the table

  • maxfield – name of column (field) to take the MAX() of

  • criteria – optional SQLAlchemy “where” criteria

Returns:

(count, maximum)

Return type:

a tuple

cardinal_pythonlib.sqlalchemy.core_query.exists_in_table(session: Session, table_: Table | TableClause, *criteria: Any) → bool[source]

Implements an efficient way of detecting if a record or records exist; should be faster than COUNT(*) in some circumstances.

Parameters:

  • session – SQLAlchemy Session, Engine (SQL Alchemy 1.4 only), or Connection object

  • table_ – SQLAlchemy Table object or table clause

  • criteria – optional SQLAlchemy “where” criteria

Returns:

a boolean

Prototypical use:

return exists_in_table(session,
                       table,
                       column(fieldname1) == value2,
                       column(fieldname2) == value2)
cardinal_pythonlib.sqlalchemy.core_query.exists_plain(session: Session, tablename: str, *criteria: Any) → bool[source]

Implements an efficient way of detecting if a record or records exist; should be faster than COUNT(*) in some circumstances.

Parameters:

  • session – SQLAlchemy Session, Engine (SQL Alchemy 1.4 only), or Connection object

  • tablename – name of the table

  • criteria – optional SQLAlchemy “where” criteria

Returns:

a boolean

Prototypical use:

return exists_plain(config.destdb.session,
                    dest_table_name,
                    column(fieldname1) == value2,
                    column(fieldname2) == value2)
cardinal_pythonlib.sqlalchemy.core_query.fetch_all_first_values(session: Session, select_statement: Select) → List[Any][source]

Returns a list of the first values in each row returned by a SELECT query.

A Core version of this sort of thing: http://xion.io/post/code/sqlalchemy-query-values.html

Parameters:

  • session – SQLAlchemy Session object

  • select_statement – SQLAlchemy Select object

Returns:

a list of the first value of each result row

cardinal_pythonlib.sqlalchemy.core_query.get_rows_fieldnames_from_raw_sql(session: Session | Engine | Connection, sql: str) → Tuple[List[Row], List[str]][source]

Returns results and column names from a query.

Parameters:

  • session – SQLAlchemy Session, Engine (SQL Alchemy 1.4 only), or Connection object

  • sql – raw SQL to execure

Returns:

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

cardinal_pythonlib.sqlalchemy.core_query.get_rows_fieldnames_from_select(session: Session | Engine | Connection, select_query: Select) → Tuple[List[Row], List[str]][source]

Returns results and column names from a query.

Parameters:

  • session – SQLAlchemy Session, Engine (SQL Alchemy 1.4 only), or Connection object

  • select_query – select() statement, i.e. instance of sqlalchemy.sql.selectable.Select

Returns:

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