element: ColumnElement[_T],
partition_by: Optional[_ByArgument] = None,
order_by: Optional[_ByArgument] = None,
- range_: Optional[typing_Tuple[Optional[typing.Any], Optional[typing.Any]]] = None,
+ range_: Optional[
+ typing_Tuple[Optional[typing.Any], Optional[typing.Any]]
+ ] = None,
rows: Optional[typing_Tuple[Optional[int], Optional[int]]] = None,
groups: Optional[typing_Tuple[Optional[int], Optional[int]]] = None,
):
)
else:
self.range_ = _FrameClause(range_) if range_ else None
- self.rows = _FrameClause(self._interpret_int_range(rows)) if rows else None
- self.groups = _FrameClause(self._interpret_int_range(groups)) if groups else None
+ self.rows = self._interpret_int_range(rows)
+ self.groups = self._interpret_int_range(groups)
if not TYPE_CHECKING:
]
)
)
-
+
def _interpret_int_range(self, rows):
+ if rows is None:
+ return None
try:
lower = rows[0] if rows[0] is None else int(rows[0])
upper = rows[1] if rows[1] is None else int(rows[1])
except ValueError as ve:
- raise exc.ArgumentError("Integer or None expected for rows value and groups value") from ve
+ err = "Integer or None expected for rows value and groups value"
+ raise exc.ArgumentError(err) from ve
- return lower, upper
+ return _FrameClause((lower, upper))
class _FrameClauseType(Enum):
lower_value, upper_value = range_
except (ValueError, TypeError) as ve:
raise exc.ArgumentError("2-tuple expected for range/rows") from ve
-
- lower_type = type_api.INTEGERTYPE if isinstance(lower_value, int) else type_api.NUMERICTYPE
- upper_type = type_api.INTEGERTYPE if isinstance(upper_value, int) else type_api.NUMERICTYPE
+
+ if isinstance(lower_value, int):
+ lower_type = type_api.INTEGERTYPE
+ else:
+ lower_type = type_api.NUMERICTYPE
+ if isinstance(upper_value, int):
+ upper_type = type_api.INTEGERTYPE
+ else:
+ upper_type = type_api.NUMERICTYPE
if lower_value is None:
self.lower_type = _FrameClauseType.RANGE_UNBOUNDED
from ... import Computed
from ... import exists
from ... import false
+from ... import Float
from ... import ForeignKey
from ... import func
from ... import Identity
from ... import TupleType
from ... import union
from ... import values
-from ... import Float
from ...exc import DatabaseError
from ...exc import ProgrammingError
@classmethod
def insert_data(cls, connection):
+ def row_factory(i):
+ return {
+ "id": i,
+ "col1": i,
+ "col2": i * 5,
+ "col3": i + 0.5,
+ }
connection.execute(
cls.tables.some_table.insert(),
- [{"id": i, "col1": i, "col2": i * 5, "col3": i + 0.5} for i in range(1, 50)],
+ [row_factory(i) for i in range(1, 50)],
)
def test_window(self, connection):