Coverage for /home/martinb/.local/share/virtualenvs/camcops/lib/python3.6/site-packages/cardinal_pythonlib/sqlalchemy/core_query.py : 40%

Hot-keys on this page
r m x p toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1#!/usr/bin/env python
2# cardinal_pythonlib/sqlalchemy/core_query.py
4"""
5===============================================================================
7 Original code copyright (C) 2009-2021 Rudolf Cardinal (rudolf@pobox.com).
9 This file is part of cardinal_pythonlib.
11 Licensed under the Apache License, Version 2.0 (the "License");
12 you may not use this file except in compliance with the License.
13 You may obtain a copy of the License at
15 https://www.apache.org/licenses/LICENSE-2.0
17 Unless required by applicable law or agreed to in writing, software
18 distributed under the License is distributed on an "AS IS" BASIS,
19 WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
20 See the License for the specific language governing permissions and
21 limitations under the License.
23===============================================================================
25**Query helper functions using the SQLAlchemy Core.**
27"""
29from typing import Any, List, Optional, Sequence, Tuple, Union
31from sqlalchemy.engine.base import Connection, Engine
32from sqlalchemy.engine.result import ResultProxy
33from sqlalchemy.orm.exc import MultipleResultsFound
34from sqlalchemy.orm.session import Session
35from sqlalchemy.sql.expression import (
36 column, exists, func, literal, select, table,
37)
38from sqlalchemy.sql.schema import Table
39from sqlalchemy.sql.selectable import Select
41from cardinal_pythonlib.logs import get_brace_style_log_with_null_handler
42from cardinal_pythonlib.sqlalchemy.dialect import SqlaDialectName
44log = get_brace_style_log_with_null_handler(__name__)
47# =============================================================================
48# Get query result with fieldnames
49# =============================================================================
51def get_rows_fieldnames_from_raw_sql(
52 session: Union[Session, Engine, Connection],
53 sql: str) -> Tuple[Sequence[Sequence[Any]], Sequence[str]]:
54 """
55 Returns results and column names from a query.
57 Args:
58 session: SQLAlchemy :class:`Session`, :class:`Engine`, or
59 :class:`Connection` object
60 sql: raw SQL to execure
62 Returns:
63 ``(rows, fieldnames)`` where ``rows`` is the usual set of results and
64 ``fieldnames`` are the name of the result columns/fields.
66 """
67 result = session.execute(sql) # type: ResultProxy
68 fieldnames = result.keys()
69 rows = result.fetchall()
70 return rows, fieldnames
73# =============================================================================
74# SELECT COUNT(*) (SQLAlchemy Core)
75# =============================================================================
76# https://stackoverflow.com/questions/12941416
78def count_star(session: Union[Session, Engine, Connection],
79 tablename: str,
80 *criteria: Any) -> int:
81 """
82 Returns the result of ``COUNT(*)`` from the specified table (with
83 additional ``WHERE`` criteria if desired).
85 Args:
86 session: SQLAlchemy :class:`Session`, :class:`Engine`, or
87 :class:`Connection` object
88 tablename: name of the table
89 criteria: optional SQLAlchemy "where" criteria
91 Returns:
92 a scalar
93 """
94 # works if you pass a connection or a session or an engine; all have
95 # the execute() method
96 query = select([func.count()]).select_from(table(tablename))
97 for criterion in criteria:
98 query = query.where(criterion)
99 return session.execute(query).scalar()
102# =============================================================================
103# SELECT COUNT(*), MAX(field) (SQLAlchemy Core)
104# =============================================================================
106def count_star_and_max(session: Union[Session, Engine, Connection],
107 tablename: str,
108 maxfield: str,
109 *criteria: Any) -> Tuple[int, Optional[int]]:
110 """
112 Args:
113 session: SQLAlchemy :class:`Session`, :class:`Engine`, or
114 :class:`Connection` object
115 tablename: name of the table
116 maxfield: name of column (field) to take the ``MAX()`` of
117 criteria: optional SQLAlchemy "where" criteria
119 Returns:
120 a tuple: ``(count, maximum)``
122 """
123 query = select([
124 func.count(),
125 func.max(column(maxfield))
126 ]).select_from(table(tablename))
127 for criterion in criteria:
128 query = query.where(criterion)
129 result = session.execute(query)
130 return result.fetchone() # count, maximum
133# =============================================================================
134# SELECT EXISTS (SQLAlchemy Core)
135# =============================================================================
136# https://stackoverflow.com/questions/15381604
137# http://docs.sqlalchemy.org/en/latest/orm/query.html
139def exists_in_table(session: Session, table_: Table, *criteria: Any) -> bool:
140 """
141 Implements an efficient way of detecting if a record or records exist;
142 should be faster than ``COUNT(*)`` in some circumstances.
144 Args:
145 session: SQLAlchemy :class:`Session`, :class:`Engine`, or
146 :class:`Connection` object
147 table_: SQLAlchemy :class:`Table` object
148 criteria: optional SQLAlchemy "where" criteria
150 Returns:
151 a boolean
153 Prototypical use:
155 .. code-block:: python
157 return exists_in_table(session,
158 table,
159 column(fieldname1) == value2,
160 column(fieldname2) == value2)
161 """
162 exists_clause = exists().select_from(table_)
163 # ... EXISTS (SELECT * FROM tablename)
164 for criterion in criteria:
165 exists_clause = exists_clause.where(criterion)
166 # ... EXISTS (SELECT * FROM tablename WHERE ...)
168 if session.get_bind().dialect.name == SqlaDialectName.MSSQL:
169 query = select([literal(True)]).where(exists_clause)
170 # ... SELECT 1 WHERE EXISTS (SELECT * FROM tablename WHERE ...)
171 else:
172 query = select([exists_clause])
173 # ... SELECT EXISTS (SELECT * FROM tablename WHERE ...)
175 result = session.execute(query).scalar()
176 return bool(result)
179def exists_plain(session: Session, tablename: str, *criteria: Any) -> bool:
180 """
181 Implements an efficient way of detecting if a record or records exist;
182 should be faster than COUNT(*) in some circumstances.
184 Args:
185 session: SQLAlchemy :class:`Session`, :class:`Engine`, or
186 :class:`Connection` object
187 tablename: name of the table
188 criteria: optional SQLAlchemy "where" criteria
190 Returns:
191 a boolean
193 Prototypical use:
195 .. code-block:: python
197 return exists_plain(config.destdb.session,
198 dest_table_name,
199 column(fieldname1) == value2,
200 column(fieldname2) == value2)
201 """
202 return exists_in_table(session, table(tablename), *criteria)
205# =============================================================================
206# Get all first values
207# =============================================================================
209def fetch_all_first_values(session: Session,
210 select_statement: Select) -> List[Any]:
211 # noinspection HttpUrlsUsage
212 """
213 Returns a list of the first values in each row returned by a ``SELECT``
214 query.
216 A Core version of this sort of thing:
217 http://xion.io/post/code/sqlalchemy-query-values.html
219 Args:
220 session: SQLAlchemy :class:`Session` object
221 select_statement: SQLAlchemy :class:`Select` object
223 Returns:
224 a list of the first value of each result row
226 """
227 rows = session.execute(select_statement) # type: ResultProxy
228 try:
229 return [row[0] for row in rows]
230 except ValueError as e:
231 raise MultipleResultsFound(str(e))