| 148 | |
| 149 | |
| 150 | class Select(Query[TableInstance, list[dict[str, Any]]]): |
| 151 | __slots__ = ( |
| 152 | "columns_list", |
| 153 | "exclude_secrets", |
| 154 | "as_of_delegate", |
| 155 | "columns_delegate", |
| 156 | "distinct_delegate", |
| 157 | "group_by_delegate", |
| 158 | "limit_delegate", |
| 159 | "offset_delegate", |
| 160 | "order_by_delegate", |
| 161 | "output_delegate", |
| 162 | "callback_delegate", |
| 163 | "where_delegate", |
| 164 | "having_delegate", |
| 165 | "lock_rows_delegate", |
| 166 | ) |
| 167 | |
| 168 | def __init__( |
| 169 | self, |
| 170 | table: type[TableInstance], |
| 171 | columns_list: Optional[Sequence[Union[Selectable, str]]] = None, |
| 172 | exclude_secrets: bool = False, |
| 173 | **kwargs, |
| 174 | ): |
| 175 | if columns_list is None: |
| 176 | columns_list = [] |
| 177 | super().__init__(table, **kwargs) |
| 178 | self.exclude_secrets = exclude_secrets |
| 179 | |
| 180 | self.as_of_delegate = AsOfDelegate() |
| 181 | self.columns_delegate = ColumnsDelegate() |
| 182 | self.distinct_delegate = DistinctDelegate() |
| 183 | self.group_by_delegate = GroupByDelegate() |
| 184 | self.limit_delegate = LimitDelegate() |
| 185 | self.offset_delegate = OffsetDelegate() |
| 186 | self.order_by_delegate = OrderByDelegate() |
| 187 | self.output_delegate = OutputDelegate() |
| 188 | self.callback_delegate = CallbackDelegate() |
| 189 | self.where_delegate = WhereDelegate() |
| 190 | self.having_delegate = WhereDelegate() |
| 191 | self.lock_rows_delegate = LockRowsDelegate() |
| 192 | |
| 193 | self.columns(*columns_list) |
| 194 | |
| 195 | def columns(self: Self, *columns: Union[Selectable, str]) -> Self: |
| 196 | _columns = self.table._process_column_args(*columns) |
| 197 | self.columns_delegate.columns(*_columns) |
| 198 | return self |
| 199 | |
| 200 | def distinct(self: Self, *, on: Optional[Sequence[Column]] = None) -> Self: |
| 201 | if on is not None and self.engine_type == "sqlite": |
| 202 | raise NotImplementedError("SQLite doesn't support DISTINCT ON") |
| 203 | |
| 204 | self.distinct_delegate.distinct(enabled=True, on=on) |
| 205 | return self |
| 206 | |
| 207 | def group_by(self: Self, *columns: Union[Column, str]) -> Self: |
no outgoing calls
no test coverage detected