Coverage for coverage / sqlitedb.py: 100.000%

1# Licensed under the Apache License: http://www.apache.org/licenses/LICENSE-2.0 

2# For details: https://github.com/coveragepy/coveragepy/blob/main/NOTICE.txt 

3 

4"""SQLite abstraction for coverage.py""" 

5 

6from __future__ import annotations 69 ctx1CDEF01GHIJ23abcdKLefghMNijklOPmn5op6QR7qr8st9ST!uv#wx$UV%yz'AB(WX)YZ4

7 

8import contextlib 69 ctx1CDEF01GHIJ23abcdKLefghMNijklOPmn5op6QR7qr8st9ST!uv#wx$UV%yz'AB(WX)YZ4

9import re 69 ctx1CDEF01GHIJ23abcdKLefghMNijklOPmn5op6QR7qr8st9ST!uv#wx$UV%yz'AB(WX)YZ4

10import sqlite3 69 ctx1CDEF01GHIJ23abcdKLefghMNijklOPmn5op6QR7qr8st9ST!uv#wx$UV%yz'AB(WX)YZ4

11from collections.abc import Iterable, Iterator 69 ctx1CDEF01GHIJ23abcdKLefghMNijklOPmn5op6QR7qr8st9ST!uv#wx$UV%yz'AB(WX)YZ4

12from typing import Any, cast 69 ctx1CDEF01GHIJ23abcdKLefghMNijklOPmn5op6QR7qr8st9ST!uv#wx$UV%yz'AB(WX)YZ4

13 

14from coverage.debug import auto_repr, clipped_repr, exc_one_line 69 ctx1CDEF01GHIJ23abcdKLefghMNijklOPmn5op6QR7qr8st9ST!uv#wx$UV%yz'AB(WX)YZ4

15from coverage.exceptions import DataError 69 ctx1CDEF01GHIJ23abcdKLefghMNijklOPmn5op6QR7qr8st9ST!uv#wx$UV%yz'AB(WX)YZ4

16from coverage.types import TDebugCtl 69 ctx1CDEF01GHIJ23abcdKLefghMNijklOPmn5op6QR7qr8st9ST!uv#wx$UV%yz'AB(WX)YZ4

17 

18 

19class SqliteDb: 69 ctx1CDEF01GHIJ23abcdKLefghMNijklOPmn5op6QR7qr8st9ST!uv#wx$UV%yz'AB(WX)YZ4

20 """A simple abstraction over a SQLite database. 

21 

22 Use as a context manager, then you can use it like a 

23 :class:`python:sqlite3.Connection` object:: 

24 

25 with SqliteDb(filename, debug_control) as db: 

26 with db.execute("select a, b from some_table") as cur: 

27 for a, b in cur: 

28 etc(a, b) 

29 

30 """ 

31 

32 def __init__(self, filename: str, debug: TDebugCtl, no_disk: bool = False) -> None: 69 ctx1CDEF01GHIJ23abcdKLefghMNijklOPmn5op6QR7qr8st9ST!uv#wx$UV%yz'AB(WX)YZ4

33 self.debug = debug 69 ctx1CDEF01GHIJ23abcdKLefghMNijklOPmn5op6QR7qr8st9ST!uv#wx$UV%yz'AB(WX)YZ4

34 self.filename = filename 69 ctx1CDEF01GHIJ23abcdKLefghMNijklOPmn5op6QR7qr8st9ST!uv#wx$UV%yz'AB(WX)YZ4

35 self.no_disk = no_disk 69 ctx1CDEF01GHIJ23abcdKLefghMNijklOPmn5op6QR7qr8st9ST!uv#wx$UV%yz'AB(WX)YZ4

36 self.nest = 0 69 ctx1CDEF01GHIJ23abcdKLefghMNijklOPmn5op6QR7qr8st9ST!uv#wx$UV%yz'AB(WX)YZ4

37 self.con: sqlite3.Connection | None = None 69 ctx1CDEF01GHIJ23abcdKLefghMNijklOPmn5op6QR7qr8st9ST!uv#wx$UV%yz'AB(WX)YZ4

38 

39 __repr__ = auto_repr 69 ctx1CDEF01GHIJ23abcdKLefghMNijklOPmn5op6QR7qr8st9ST!uv#wx$UV%yz'AB(WX)YZ4

40 

41 def _connect(self) -> None: 69 ctx1CDEF01GHIJ23abcdKLefghMNijklOPmn5op6QR7qr8st9ST!uv#wx$UV%yz'AB(WX)YZ4

42 """Connect to the db and do universal initialization.""" 

43 if self.con is not None: 69 ctx1CDEF01GHIJ23abcdKLefghMNijklOPmn5op6QR7qr8st9ST!uv#wx$UV%yz'AB(WX)YZ4

44 return 57 ctx1CDEF01GHIJ23abcdKLefghMNijklOPmnopQRqrstSTuvwxUVyzABWXYZ4

45 

46 # It can happen that Python switches threads while the tracer writes 

47 # data. The second thread will also try to write to the data, 

48 # effectively causing a nested context. However, given the idempotent 

49 # nature of the tracer operations, sharing a connection among threads 

50 # is not a problem. 

51 if self.debug.should("sql"): 69 ctx1CDEF01GHIJ23abcdKLefghMNijklOPmn5op6QR7qr8st9ST!uv#wx$UV%yz'AB(WX)YZ4

52 self.debug.write(f"Connecting to {self.filename!r}") 57 ctx1CDEF01GHIJ23abcdKLefghMNijklOPmnopQRqrstSTuvwxUVyzABWXYZ4

53 try: 69 ctx1CDEF01GHIJ23abcdKLefghMNijklOPmn5op6QR7qr8st9ST!uv#wx$UV%yz'AB(WX)YZ4

54 # Use uri=True when connecting to memory URIs 

55 if self.filename.startswith("file:"): 69 ctx1CDEF01GHIJ23abcdKLefghMNijklOPmn5op6QR7qr8st9ST!uv#wx$UV%yz'AB(WX)YZ4

56 self.con = sqlite3.connect(self.filename, check_same_thread=False, uri=True) 57 ctx1CDEF01GHIJ23abcdKLefghMNijklOPmnopQRqrstSTuvwxUVyzABWXYZ4

57 else: 

58 self.con = sqlite3.connect(self.filename, check_same_thread=False) 69 ctx1CDEF01GHIJ23abcdKLefghMNijklOPmn5op6QR7qr8st9ST!uv#wx$UV%yz'AB(WX)YZ4

59 except sqlite3.Error as exc: 38 ctx1CDEFGHIJabcdefghijklmnopqrstuvwxyzABYZ

60 raise DataError(f"Couldn't use data file {self.filename!r}: {exc}") from exc 38 ctx1CDEFGHIJabcdefghijklmnopqrstuvwxyzABYZ

61 

62 if self.debug.should("sql"): 69 ctx1CDEF01GHIJ23abcdKLefghMNijklOPmn5op6QR7qr8st9ST!uv#wx$UV%yz'AB(WX)YZ4

63 self.debug.write(f"Connected to {self.filename!r} as {self.con!r}") 57 ctx1CDEF01GHIJ23abcdKLefghMNijklOPmnopQRqrstSTuvwxUVyzABWXYZ4

64 

65 self.con.create_function("REGEXP", 2, lambda txt, pat: re.search(txt, pat) is not None) 69 ctx1CDEF01GHIJ23abcdKLefghMNijklOPmn5op6QR7qr8st9ST!uv#wx$UV%yz'AB(WX)YZ4

66 

67 # Turning off journal_mode can speed up writing. It can't always be 

68 # disabled, so we have to be prepared for *-journal files elsewhere. 

69 # In Python 3.12+, we can change the config to allow journal_mode=off. 

70 if hasattr(sqlite3, "SQLITE_DBCONFIG_DEFENSIVE"): 69 ctx1CDEF01GHIJ23abcdKLefghMNijklOPmn5op6QR7qr8st9ST!uv#wx$UV%yz'AB(WX)YZ4

71 # Turn off defensive mode, so that journal_mode=off can succeed. 

72 self.con.setconfig( # type: ignore[attr-defined, unused-ignore] 54 ctx1abcdKLefghMNijklOPmn5op6QR7qr8st9ST!uv#wx$UV%yz'AB(WX)

73 sqlite3.SQLITE_DBCONFIG_DEFENSIVE, 

74 False, 

75 ) 

76 

77 # This pragma makes writing faster. It disables rollbacks, but we never need them. 

78 self.execute_void("pragma journal_mode=off") 69 ctx1CDEF01GHIJ23abcdKLefghMNijklOPmn5op6QR7qr8st9ST!uv#wx$UV%yz'AB(WX)YZ4

79 

80 # This pragma makes writing faster. It can fail in unusual situations 

81 # (https://github.com/coveragepy/coveragepy/issues/1646), so use fail_ok=True 

82 # to keep things going. 

83 self.execute_void("pragma synchronous=off", fail_ok=True) 69 ctx1CDEF01GHIJ23abcdKLefghMNijklOPmn5op6QR7qr8st9ST!uv#wx$UV%yz'AB(WX)YZ4

84 

85 def close(self, force: bool = False) -> None: 69 ctx1CDEF01GHIJ23abcdKLefghMNijklOPmn5op6QR7qr8st9ST!uv#wx$UV%yz'AB(WX)YZ4

86 """If needed, close the connection.""" 

87 if self.con is not None: 69 ctx1CDEF01GHIJ23abcdKLefghMNijklOPmn5op6QR7qr8st9ST!uv#wx$UV%yz'AB(WX)YZ4

88 if force or not self.no_disk: 69 ctx1CDEF01GHIJ23abcdKLefghMNijklOPmn5op6QR7qr8st9ST!uv#wx$UV%yz'AB(WX)YZ4

89 if self.debug.should("sql"): 69 ctx1CDEF01GHIJ23abcdKLefghMNijklOPmn5op6QR7qr8st9ST!uv#wx$UV%yz'AB(WX)YZ4

90 self.debug.write(f"Closing {self.con!r} on {self.filename!r}") 57 ctx1CDEF01GHIJ23abcdKLefghMNijklOPmnopQRqrstSTuvwxUVyzABWXYZ4

91 self.con.close() 69 ctx1CDEF01GHIJ23abcdKLefghMNijklOPmn5op6QR7qr8st9ST!uv#wx$UV%yz'AB(WX)YZ4

92 self.con = None 69 ctx1CDEF01GHIJ23abcdKLefghMNijklOPmn5op6QR7qr8st9ST!uv#wx$UV%yz'AB(WX)YZ4

93 

94 def __enter__(self) -> SqliteDb: 69 ctx1CDEF01GHIJ23abcdKLefghMNijklOPmn5op6QR7qr8st9ST!uv#wx$UV%yz'AB(WX)YZ4

95 if self.nest == 0: 69 ctx1CDEF01GHIJ23abcdKLefghMNijklOPmn5op6QR7qr8st9ST!uv#wx$UV%yz'AB(WX)YZ4

96 self._connect() 69 ctx1CDEF01GHIJ23abcdKLefghMNijklOPmn5op6QR7qr8st9ST!uv#wx$UV%yz'AB(WX)YZ4

97 assert self.con is not None 69 ctx1CDEF01GHIJ23abcdKLefghMNijklOPmn5op6QR7qr8st9ST!uv#wx$UV%yz'AB(WX)YZ4

98 self.con.__enter__() 69 ctx1CDEF01GHIJ23abcdKLefghMNijklOPmn5op6QR7qr8st9ST!uv#wx$UV%yz'AB(WX)YZ4

99 self.nest += 1 69 ctx1CDEF01GHIJ23abcdKLefghMNijklOPmn5op6QR7qr8st9ST!uv#wx$UV%yz'AB(WX)YZ4

100 return self 69 ctx1CDEF01GHIJ23abcdKLefghMNijklOPmn5op6QR7qr8st9ST!uv#wx$UV%yz'AB(WX)YZ4

101 

102 def __exit__(self, exc_type, exc_value, traceback) -> None: # type: ignore[no-untyped-def] 69 ctx1CDEF01GHIJ23abcdKLefghMNijklOPmn5op6QR7qr8st9ST!uv#wx$UV%yz'AB(WX)YZ4

103 self.nest -= 1 69 ctx1CDEF01GHIJ23abcdKLefghMNijklOPmn5op6QR7qr8st9ST!uv#wx$UV%yz'AB(WX)YZ4

104 if self.nest == 0: 69 ctx1CDEF01GHIJ23abcdKLefghMNijklOPmn5op6QR7qr8st9ST!uv#wx$UV%yz'AB(WX)YZ4

105 try: 69 ctx1CDEF01GHIJ23abcdKLefghMNijklOPmn5op6QR7qr8st9ST!uv#wx$UV%yz'AB(WX)YZ4

106 assert self.con is not None 69 ctx1CDEF01GHIJ23abcdKLefghMNijklOPmn5op6QR7qr8st9ST!uv#wx$UV%yz'AB(WX)YZ4

107 self.con.__exit__(exc_type, exc_value, traceback) 69 ctx1CDEF01GHIJ23abcdKLefghMNijklOPmn5op6QR7qr8st9ST!uv#wx$UV%yz'AB(WX)YZ4

108 self.close() 69 ctx1CDEF01GHIJ23abcdKLefghMNijklOPmn5op6QR7qr8st9ST!uv#wx$UV%yz'AB(WX)YZ4

109 except Exception as exc: 57 ctx1CDEF01GHIJ23abcdKLefghMNijklOPmnopQRqrstSTuvwxUVyzABWXYZ4

110 if self.debug.should("sql"): 57 ctx1CDEF01GHIJ23abcdKLefghMNijklOPmnopQRqrstSTuvwxUVyzABWXYZ4

111 self.debug.write(f"EXCEPTION from __exit__: {exc_one_line(exc)}") 57 ctx1CDEF01GHIJ23abcdKLefghMNijklOPmnopQRqrstSTuvwxUVyzABWXYZ4

112 raise DataError(f"Couldn't end data file {self.filename!r}: {exc}") from exc 57 ctx1CDEF01GHIJ23abcdKLefghMNijklOPmnopQRqrstSTuvwxUVyzABWXYZ4

113 

114 def _execute(self, sql: str, parameters: Iterable[Any]) -> sqlite3.Cursor: 69 ctx1CDEF01GHIJ23abcdKLefghMNijklOPmn5op6QR7qr8st9ST!uv#wx$UV%yz'AB(WX)YZ4

115 """Same as :meth:`python:sqlite3.Connection.execute`.""" 

116 if self.debug.should("sql"): 69 ctx1CDEF01GHIJ23abcdKLefghMNijklOPmn5op6QR7qr8st9ST!uv#wx$UV%yz'AB(WX)YZ4

117 tail = f" with {parameters!r}" if parameters else "" 57 ctx1CDEF01GHIJ23abcdKLefghMNijklOPmnopQRqrstSTuvwxUVyzABWXYZ4

118 self.debug.write(f"Executing {sql!r}{tail}") 57 ctx1CDEF01GHIJ23abcdKLefghMNijklOPmnopQRqrstSTuvwxUVyzABWXYZ4

119 try: 69 ctx1CDEF01GHIJ23abcdKLefghMNijklOPmn5op6QR7qr8st9ST!uv#wx$UV%yz'AB(WX)YZ4

120 assert self.con is not None 69 ctx1CDEF01GHIJ23abcdKLefghMNijklOPmn5op6QR7qr8st9ST!uv#wx$UV%yz'AB(WX)YZ4

121 try: 69 ctx1CDEF01GHIJ23abcdKLefghMNijklOPmn5op6QR7qr8st9ST!uv#wx$UV%yz'AB(WX)YZ4

122 return self.con.execute(sql, parameters) # type: ignore[arg-type] 69 ctx1CDEF01GHIJ23abcdKLefghMNijklOPmn5op6QR7qr8st9ST!uv#wx$UV%yz'AB(WX)YZ4

123 except Exception: 69 ctx1CDEF01GHIJ23abcdKLefghMNijklOPmn5op6QR7qr8st9ST!uv#wx$UV%yz'AB(WX)YZ4

124 # In some cases, an error might happen that isn't really an 

125 # error. Try again immediately. 

126 # https://github.com/coveragepy/coveragepy/issues/1010 

127 return self.con.execute(sql, parameters) # type: ignore[arg-type] 69 ctx1CDEF01GHIJ23abcdKLefghMNijklOPmn5op6QR7qr8st9ST!uv#wx$UV%yz'AB(WX)YZ4

128 except sqlite3.Error as exc: 69 ctx1CDEF01GHIJ23abcdKLefghMNijklOPmn5op6QR7qr8st9ST!uv#wx$UV%yz'AB(WX)YZ4

129 msg = str(exc) 69 ctx1CDEF01GHIJ23abcdKLefghMNijklOPmn5op6QR7qr8st9ST!uv#wx$UV%yz'AB(WX)YZ4

130 if not self.no_disk: 69 ctx1CDEF01GHIJ23abcdKLefghMNijklOPmn5op6QR7qr8st9ST!uv#wx$UV%yz'AB(WX)YZ4

131 try: 69 ctx1CDEF01GHIJ23abcdKLefghMNijklOPmn5op6QR7qr8st9ST!uv#wx$UV%yz'AB(WX)YZ4

132 # `execute` is the first thing we do with the database, so try 

133 # hard to provide useful hints if something goes wrong now. 

134 with open(self.filename, "rb") as bad_file: 69 ctx1CDEF01GHIJ23abcdKLefghMNijklOPmn5op6QR7qr8st9ST!uv#wx$UV%yz'AB(WX)YZ4

135 cov4_sig = b"!coverage.py: This is a private format" 69 ctx1CDEF01GHIJ23abcdKLefghMNijklOPmn5op6QR7qr8st9ST!uv#wx$UV%yz'AB(WX)YZ4

136 if bad_file.read(len(cov4_sig)) == cov4_sig: 69 ctx1CDEF01GHIJ23abcdKLefghMNijklOPmn5op6QR7qr8st9ST!uv#wx$UV%yz'AB(WX)YZ4

137 msg = ( 57 ctx1CDEF01GHIJ23abcdKLefghMNijklOPmnopQRqrstSTuvwxUVyzABWXYZ4

138 "Looks like a coverage 4.x data file. " 

139 + "Are you mixing versions of coverage?" 

140 ) 

141 except Exception: 57 ctx1CDEF01GHIJ23abcdKLefghMNijklOPmnopQRqrstSTuvwxUVyzABWXYZ4

142 pass 57 ctx1CDEF01GHIJ23abcdKLefghMNijklOPmnopQRqrstSTuvwxUVyzABWXYZ4

143 if self.debug.should("sql"): 69 ctx1CDEF01GHIJ23abcdKLefghMNijklOPmn5op6QR7qr8st9ST!uv#wx$UV%yz'AB(WX)YZ4

144 self.debug.write(f"EXCEPTION from execute: {exc_one_line(exc)}") 57 ctx1CDEF01GHIJ23abcdKLefghMNijklOPmnopQRqrstSTuvwxUVyzABWXYZ4

145 raise DataError(f"Couldn't use data file {self.filename!r}: {msg}") from exc 69 ctx1CDEF01GHIJ23abcdKLefghMNijklOPmn5op6QR7qr8st9ST!uv#wx$UV%yz'AB(WX)YZ4

146 

147 @contextlib.contextmanager 69 ctx1CDEF01GHIJ23abcdKLefghMNijklOPmn5op6QR7qr8st9ST!uv#wx$UV%yz'AB(WX)YZ4

148 def execute( 66 ctx1CDEF01GHIJ23abcdKLefghMNijklOPmn5op6QR7qr8st9ST!uv#wx$UV%yz'AB(WX)

149 self, 

150 sql: str, 

151 parameters: Iterable[Any] = (), 

152 ) -> Iterator[sqlite3.Cursor]: 

153 """Context managed :meth:`python:sqlite3.Connection.execute`. 

154 

155 Use with a ``with`` statement to auto-close the returned cursor. 

156 """ 

157 cur = self._execute(sql, parameters) 69 ctx1CDEF01GHIJ23abcdKLefghMNijklOPmn5op6QR7qr8st9ST!uv#wx$UV%yz'AB(WX)YZ4

158 try: 69 ctx1CDEF01GHIJ23abcdKLefghMNijklOPmn5op6QR7qr8st9ST!uv#wx$UV%yz'AB(WX)YZ4

159 yield cur 69 ctx1CDEF01GHIJ23abcdKLefghMNijklOPmn5op6QR7qr8st9ST!uv#wx$UV%yz'AB(WX)YZ4

160 finally: 

161 cur.close() 69 ctx1CDEF01GHIJ23abcdKLefghMNijklOPmn5op6QR7qr8st9ST!uv#wx$UV%yz'AB(WX)YZ4

162 

163 def execute_void(self, sql: str, parameters: Iterable[Any] = (), fail_ok: bool = False) -> None: 69 ctx1CDEF01GHIJ23abcdKLefghMNijklOPmn5op6QR7qr8st9ST!uv#wx$UV%yz'AB(WX)YZ4

164 """Same as :meth:`python:sqlite3.Connection.execute` when you don't need the cursor. 

165 

166 If `fail_ok` is True, then SQLite errors are ignored. 

167 """ 

168 try: 69 ctx1CDEF01GHIJ23abcdKLefghMNijklOPmn5op6QR7qr8st9ST!uv#wx$UV%yz'AB(WX)YZ4

169 # PyPy needs the .close() calls here, or sqlite gets twisted up: 

170 # https://bitbucket.org/pypy/pypy/issues/2872/default-isolation-mode-is-different-on 

171 self._execute(sql, parameters).close() 69 ctx1CDEF01GHIJ23abcdKLefghMNijklOPmn5op6QR7qr8st9ST!uv#wx$UV%yz'AB(WX)YZ4

172 except DataError: 57 ctx1CDEF01GHIJ23abcdKLefghMNijklOPmnopQRqrstSTuvwxUVyzABWXYZ4

173 if not fail_ok: 57 ctx1CDEF01GHIJ23abcdKLefghMNijklOPmnopQRqrstSTuvwxUVyzABWXYZ4

174 raise 57 ctx1CDEF01GHIJ23abcdKLefghMNijklOPmnopQRqrstSTuvwxUVyzABWXYZ4

175 

176 def execute_for_rowid(self, sql: str, parameters: Iterable[Any] = ()) -> int: 69 ctx1CDEF01GHIJ23abcdKLefghMNijklOPmn5op6QR7qr8st9ST!uv#wx$UV%yz'AB(WX)YZ4

177 """Like execute, but returns the lastrowid.""" 

178 with self.execute(sql, parameters) as cur: 69 ctx1CDEF01GHIJ23abcdKLefghMNijklOPmn5op6QR7qr8st9ST!uv#wx$UV%yz'AB(WX)YZ4

179 assert cur.lastrowid is not None 69 ctx1CDEF01GHIJ23abcdKLefghMNijklOPmn5op6QR7qr8st9ST!uv#wx$UV%yz'AB(WX)YZ4

180 rowid: int = cur.lastrowid 69 ctx1CDEF01GHIJ23abcdKLefghMNijklOPmn5op6QR7qr8st9ST!uv#wx$UV%yz'AB(WX)YZ4

181 if self.debug.should("sqldata"): 69 ctx1CDEF01GHIJ23abcdKLefghMNijklOPmn5op6QR7qr8st9ST!uv#wx$UV%yz'AB(WX)YZ4

182 self.debug.write(f"Row id result: {rowid!r}") 57 ctx1CDEF01GHIJ23abcdKLefghMNijklOPmnopQRqrstSTuvwxUVyzABWXYZ4

183 return rowid 69 ctx1CDEF01GHIJ23abcdKLefghMNijklOPmn5op6QR7qr8st9ST!uv#wx$UV%yz'AB(WX)YZ4

184 

185 def execute_one(self, sql: str, parameters: Iterable[Any] = ()) -> tuple[Any, ...] | None: 69 ctx1CDEF01GHIJ23abcdKLefghMNijklOPmn5op6QR7qr8st9ST!uv#wx$UV%yz'AB(WX)YZ4

186 """Execute a statement and return the one row that results. 

187 

188 This is like execute(sql, parameters).fetchone(), except it is 

189 correct in reading the entire result set. This will raise an 

190 exception if more than one row results. 

191 

192 Returns a row, or None if there were no rows. 

193 """ 

194 with self.execute(sql, parameters) as cur: 69 ctx1CDEF01GHIJ23abcdKLefghMNijklOPmn5op6QR7qr8st9ST!uv#wx$UV%yz'AB(WX)YZ4

195 rows = list(cur) 69 ctx1CDEF01GHIJ23abcdKLefghMNijklOPmn5op6QR7qr8st9ST!uv#wx$UV%yz'AB(WX)YZ4

196 if len(rows) == 0: 69 ctx1CDEF01GHIJ23abcdKLefghMNijklOPmn5op6QR7qr8st9ST!uv#wx$UV%yz'AB(WX)YZ4

197 return None 69 ctx1CDEF01GHIJ23abcdKLefghMNijklOPmn5op6QR7qr8st9ST!uv#wx$UV%yz'AB(WX)YZ4

198 elif len(rows) == 1: 69 ctx1CDEF01GHIJ23abcdKLefghMNijklOPmn5op6QR7qr8st9ST!uv#wx$UV%yz'AB(WX)YZ4

199 return cast(tuple[Any, ...], rows[0]) 69 ctx1CDEF01GHIJ23abcdKLefghMNijklOPmn5op6QR7qr8st9ST!uv#wx$UV%yz'AB(WX)YZ4

200 else: 

201 raise AssertionError(f"SQL {sql!r} shouldn't return {len(rows)} rows") 

202 

203 def _executemany(self, sql: str, data: list[Any]) -> sqlite3.Cursor: 69 ctx1CDEF01GHIJ23abcdKLefghMNijklOPmn5op6QR7qr8st9ST!uv#wx$UV%yz'AB(WX)YZ4

204 """Same as :meth:`python:sqlite3.Connection.executemany`.""" 

205 if self.debug.should("sql"): 69 ctx1CDEF01GHIJ23abcdKLefghMNijklOPmn5op6QR7qr8st9ST!uv#wx$UV%yz'AB(WX)YZ4

206 final = ":" if self.debug.should("sqldata") else "" 57 ctx1CDEF01GHIJ23abcdKLefghMNijklOPmnopQRqrstSTuvwxUVyzABWXYZ4

207 self.debug.write(f"Executing many {sql!r} with {len(data)} rows{final}") 57 ctx1CDEF01GHIJ23abcdKLefghMNijklOPmnopQRqrstSTuvwxUVyzABWXYZ4

208 if self.debug.should("sqldata"): 57 ctx1CDEF01GHIJ23abcdKLefghMNijklOPmnopQRqrstSTuvwxUVyzABWXYZ4

209 for i, row in enumerate(data): 57 ctx1CDEF01GHIJ23abcdKLefghMNijklOPmnopQRqrstSTuvwxUVyzABWXYZ4

210 self.debug.write(f"{i:4d}: {row!r}") 57 ctx1CDEF01GHIJ23abcdKLefghMNijklOPmnopQRqrstSTuvwxUVyzABWXYZ4

211 assert self.con is not None 69 ctx1CDEF01GHIJ23abcdKLefghMNijklOPmn5op6QR7qr8st9ST!uv#wx$UV%yz'AB(WX)YZ4

212 try: 69 ctx1CDEF01GHIJ23abcdKLefghMNijklOPmn5op6QR7qr8st9ST!uv#wx$UV%yz'AB(WX)YZ4

213 return self.con.executemany(sql, data) 69 ctx1CDEF01GHIJ23abcdKLefghMNijklOPmn5op6QR7qr8st9ST!uv#wx$UV%yz'AB(WX)YZ4

214 except Exception: 57 ctx1CDEF01GHIJ23abcdKLefghMNijklOPmnopQRqrstSTuvwxUVyzABWXYZ4

215 # In some cases, an error might happen that isn't really an 

216 # error. Try again immediately. 

217 # https://github.com/coveragepy/coveragepy/issues/1010 

218 return self.con.executemany(sql, data) 57 ctx1CDEF01GHIJ23abcdKLefghMNijklOPmnopQRqrstSTuvwxUVyzABWXYZ4

219 

220 def executemany_void(self, sql: str, data: list[Any]) -> None: 69 ctx1CDEF01GHIJ23abcdKLefghMNijklOPmn5op6QR7qr8st9ST!uv#wx$UV%yz'AB(WX)YZ4

221 """Same as :meth:`python:sqlite3.Connection.executemany` when you don't need the cursor.""" 

222 self._executemany(sql, data).close() 69 ctx1CDEF01GHIJ23abcdKLefghMNijklOPmn5op6QR7qr8st9ST!uv#wx$UV%yz'AB(WX)YZ4

223 

224 def executescript(self, script: str) -> None: 69 ctx1CDEF01GHIJ23abcdKLefghMNijklOPmn5op6QR7qr8st9ST!uv#wx$UV%yz'AB(WX)YZ4

225 """Same as :meth:`python:sqlite3.Connection.executescript`.""" 

226 if self.debug.should("sql"): 69 ctx1CDEF01GHIJ23abcdKLefghMNijklOPmn5op6QR7qr8st9ST!uv#wx$UV%yz'AB(WX)YZ4

227 self.debug.write( 57 ctx1CDEF01GHIJ23abcdKLefghMNijklOPmnopQRqrstSTuvwxUVyzABWXYZ4

228 "Executing script with {} chars: {}".format( 

229 len(script), 

230 clipped_repr(script, 100), 

231 ) 

232 ) 

233 assert self.con is not None 69 ctx1CDEF01GHIJ23abcdKLefghMNijklOPmn5op6QR7qr8st9ST!uv#wx$UV%yz'AB(WX)YZ4

234 self.con.executescript(script).close() 69 ctx1CDEF01GHIJ23abcdKLefghMNijklOPmn5op6QR7qr8st9ST!uv#wx$UV%yz'AB(WX)YZ4

235 

236 def dump(self) -> str: 69 ctx1CDEF01GHIJ23abcdKLefghMNijklOPmn5op6QR7qr8st9ST!uv#wx$UV%yz'AB(WX)YZ4

237 """Return a multi-line string, the SQL dump of the database.""" 

238 assert self.con is not None 69 ctx1CDEF01GHIJ23abcdKLefghMNijklOPmn5op6QR7qr8st9ST!uv#wx$UV%yz'AB(WX)YZ4

239 return "\n".join(self.con.iterdump()) 69 ctx1CDEF01GHIJ23abcdKLefghMNijklOPmn5op6QR7qr8st9ST!uv#wx$UV%yz'AB(WX)YZ4