Hide keyboard shortcuts

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 

3 

4""" 

5=============================================================================== 

6 

7 Original code copyright (C) 2009-2021 Rudolf Cardinal (rudolf@pobox.com). 

8 

9 This file is part of cardinal_pythonlib. 

10 

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 

14 

15 https://www.apache.org/licenses/LICENSE-2.0 

16 

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. 

22 

23=============================================================================== 

24 

25**Functions to operate with the raw Python database API.** 

26 

27See https://www.python.org/dev/peps/pep-0249/. 

28 

29""" 

30 

31from collections import OrderedDict 

32from typing import Any, Dict, Generator, List, Optional 

33 

34from cardinal_pythonlib.typing_helpers import Pep249DatabaseCursorType as Cursor # noqa 

35 

36 

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] 

42 

43 

44def genrows(cursor: Cursor, arraysize: int = 1000) \ 

45 -> Generator[List[Any], None, None]: 

46 """ 

47 Generate all rows from a cursor. 

48 

49 Args: 

50 cursor: the cursor 

51 arraysize: split fetches into chunks of this many records 

52 

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 

63 

64 

65def genfirstvalues(cursor: Cursor, arraysize: int = 1000) \ 

66 -> Generator[Any, None, None]: 

67 """ 

68 Generate the first value in each row. 

69 

70 Args: 

71 cursor: the cursor 

72 arraysize: split fetches into chunks of this many records 

73 

74 Yields: 

75 the first value of each row 

76 """ 

77 return (row[0] for row in genrows(cursor, arraysize)) 

78 

79 

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()] 

85 

86 

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. 

91 

92 Args: 

93 cursor: the cursor 

94 arraysize: split fetches into chunks of this many records 

95 

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 ) 

105 

106 

107def dictfetchall(cursor: Cursor) -> List[Dict[str, Any]]: 

108 """ 

109 Return all rows from a cursor as a list of :class:`OrderedDict` objects. 

110 

111 Args: 

112 cursor: the cursor 

113 

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 ] 

123 

124 

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))