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

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/dbfunc.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**Functions to operate with the raw Python database API.**
27See https://www.python.org/dev/peps/pep-0249/.
29"""
31from collections import OrderedDict
32from typing import Any, Dict, Generator, List, Optional
34from cardinal_pythonlib.typing_helpers import Pep249DatabaseCursorType as Cursor # noqa
37def get_fieldnames_from_cursor(cursor: Cursor) -> List[str]:
38 """
39 Get a list of fieldnames from an executed cursor.
40 """
41 return [i[0] for i in cursor.description]
44def genrows(cursor: Cursor, arraysize: int = 1000) \
45 -> Generator[List[Any], None, None]:
46 """
47 Generate all rows from a cursor.
49 Args:
50 cursor: the cursor
51 arraysize: split fetches into chunks of this many records
53 Yields:
54 each row
55 """
56 # http://code.activestate.com/recipes/137270-use-generators-for-fetching-large-db-record-sets/ # noqa
57 while True:
58 results = cursor.fetchmany(arraysize)
59 if not results:
60 break
61 for result in results:
62 yield result
65def genfirstvalues(cursor: Cursor, arraysize: int = 1000) \
66 -> Generator[Any, None, None]:
67 """
68 Generate the first value in each row.
70 Args:
71 cursor: the cursor
72 arraysize: split fetches into chunks of this many records
74 Yields:
75 the first value of each row
76 """
77 return (row[0] for row in genrows(cursor, arraysize))
80def fetchallfirstvalues(cursor: Cursor) -> List[Any]:
81 """
82 Return a list of the first value in each row.
83 """
84 return [row[0] for row in cursor.fetchall()]
87def gendicts(cursor: Cursor, arraysize: int = 1000) \
88 -> Generator[Dict[str, Any], None, None]:
89 """
90 Generate all rows from a cursor as :class:`OrderedDict` objects.
92 Args:
93 cursor: the cursor
94 arraysize: split fetches into chunks of this many records
96 Yields:
97 each row, as an :class:`OrderedDict` whose key are column names
98 and whose values are the row values
99 """
100 columns = get_fieldnames_from_cursor(cursor)
101 return (
102 OrderedDict(zip(columns, row))
103 for row in genrows(cursor, arraysize)
104 )
107def dictfetchall(cursor: Cursor) -> List[Dict[str, Any]]:
108 """
109 Return all rows from a cursor as a list of :class:`OrderedDict` objects.
111 Args:
112 cursor: the cursor
114 Returns:
115 a list (one item per row) of :class:`OrderedDict` objects whose key are
116 column names and whose values are the row values
117 """
118 columns = get_fieldnames_from_cursor(cursor)
119 return [
120 OrderedDict(zip(columns, row))
121 for row in cursor.fetchall()
122 ]
125def dictfetchone(cursor: Cursor) -> Optional[Dict[str, Any]]:
126 """
127 Return the next row from a cursor as an :class:`OrderedDict`, or ``None``.
128 """
129 columns = get_fieldnames_from_cursor(cursor)
130 row = cursor.fetchone()
131 if not row:
132 return None
133 return OrderedDict(zip(columns, row))