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.

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, 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, 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, *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, or Connection object

  • table_ – SQLAlchemy Table object

  • 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, 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[Sequence[Sequence[Any]], Sequence[str]][source]

Returns results and column names from a query.

Parameters:

  • session – SQLAlchemy Session, Engine, 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.