(self, order_by=None)
| 259 | return params |
| 260 | |
| 261 | def select_all(self, order_by=None): |
| 262 | # type: (Union[str, Sequence[str], None]) -> Iterator[Any] |
| 263 | key = 'select-all' |
| 264 | if order_by: |
| 265 | if isinstance(order_by, str): |
| 266 | order_by = [order_by] |
| 267 | elif isinstance(order_by, (list, tuple)): |
| 268 | order_by = order_by |
| 269 | else: |
| 270 | raise ValueError(f'select_all: \"order_by\" invalid type.') |
| 271 | for column_name in order_by: |
| 272 | if not isinstance(column_name, str): |
| 273 | raise ValueError(f'select_all: \"order_by\" invalid type.') |
| 274 | if column_name.lower() not in self.schema.class_fields: |
| 275 | raise ValueError( |
| 276 | f'select_all: table \"{self.schema.table_name}\" does not have column \"{column_name}\"') |
| 277 | key += f': ' + ', '.join(order_by) |
| 278 | |
| 279 | query = self._queries.get(key) |
| 280 | if not query: |
| 281 | query = 'SELECT ' + ', '.join(self.schema.columns) |
| 282 | query += f' FROM {self.schema.table_name}' |
| 283 | if self.schema.owner_column: |
| 284 | query += f' WHERE {self.schema.owner_column}=:{self.schema.owner_column}' |
| 285 | if order_by: |
| 286 | query += ' ORDER BY ' + ', '.join(order_by) |
| 287 | self._queries[key] = query |
| 288 | |
| 289 | conn = self.get_connection() |
| 290 | if self.schema.owner_column: |
| 291 | params = {self.schema.owner_column: self.owner} |
| 292 | curr = conn.execute(query, params) |
| 293 | else: |
| 294 | curr = conn.execute(query) |
| 295 | for row in curr: |
| 296 | yield self._populate_data_object(row) |
| 297 | |
| 298 | def select_by_filter(self, columns, values): |
| 299 | # type: (Union[str, Sequence[str]], Union[Any, Sequence[Any]]) -> Iterable[Any] |
no test coverage detected