diff --git a/bmsdna/lakeapi/context/df_base.py b/bmsdna/lakeapi/context/df_base.py index 868da21..b36e547 100644 --- a/bmsdna/lakeapi/context/df_base.py +++ b/bmsdna/lakeapi/context/df_base.py @@ -384,6 +384,7 @@ def register_datasource( file_type: FileTypes, filters: Optional[FilterType], meta_only: bool = False, + limit: int | None = None, ): ds = self.get_pyarrow_dataset(uri, file_type, filters) self.modified_dates[target_name] = self.get_modified_date(uri, file_type) diff --git a/bmsdna/lakeapi/context/df_duckdb.py b/bmsdna/lakeapi/context/df_duckdb.py index dda7bbb..850c09d 100644 --- a/bmsdna/lakeapi/context/df_duckdb.py +++ b/bmsdna/lakeapi/context/df_duckdb.py @@ -343,6 +343,7 @@ def register_datasource( file_type: FileTypes, filters: Optional[FilterType], meta_only: bool = False, + limit: int | None = None, ): self.modified_dates[target_name] = self.get_modified_date(uri, file_type) @@ -390,6 +391,7 @@ def register_datasource( storage_options=uri_opts, conditions=filters, use_fsspec=os.getenv("DUCKDB_DELTA_USE_FSSPEC", "0") == "1", + limit=limit, ) return if file_type == "duckdb": diff --git a/bmsdna/lakeapi/context/df_odbc.py b/bmsdna/lakeapi/context/df_odbc.py index 20e699c..94251cd 100644 --- a/bmsdna/lakeapi/context/df_odbc.py +++ b/bmsdna/lakeapi/context/df_odbc.py @@ -186,6 +186,7 @@ def register_datasource( file_type: FileTypes, filters: Any, meta_only: bool = False, + limit: int | None = None, ): assert file_type == "odbc" assert uri.account is None diff --git a/bmsdna/lakeapi/context/df_polars.py b/bmsdna/lakeapi/context/df_polars.py index 0789278..6bc84fc 100644 --- a/bmsdna/lakeapi/context/df_polars.py +++ b/bmsdna/lakeapi/context/df_polars.py @@ -231,6 +231,7 @@ def register_datasource( file_type: FileTypes, filters: Optional[FilterType], meta_only: bool = False, + limit: int | None = None, ): import polars as pl @@ -245,7 +246,7 @@ def register_datasource( db_uri, db_opts = uri.get_uri_options(flavor="original") from deltalake2db import polars_scan_delta, get_polars_schema - if meta_only: + if meta_only or limit == 0: schema = get_polars_schema(db_uri, storage_options=db_opts) df = pl.DataFrame(schema=schema) else: @@ -253,6 +254,7 @@ def register_datasource( db_uri, storage_options=db_opts, conditions=filters, + limit=limit, ) except DeltaProtocolError as de: raise FileTypeNotSupportedError( diff --git a/bmsdna/lakeapi/core/config.py b/bmsdna/lakeapi/core/config.py index b472381..a3fe54c 100644 --- a/bmsdna/lakeapi/core/config.py +++ b/bmsdna/lakeapi/core/config.py @@ -160,6 +160,12 @@ class DuckDBBackendConfig: index: List[str] | None = None +def _to_dict(d: Union[list, dict]): + if isinstance(d, list): + return {i["key"]: i["value"] for i in d} + return d + + @dataclass class Config: name: str @@ -225,7 +231,7 @@ def _from_dict(cls, config: Dict, basic_config: BasicConfig, accounts: dict): meta = get_deltalake_meta(uri_obj) assert meta.last_metadata is not None cfg = json.loads( - meta.last_metadata.get("configuration", {}).get( + _to_dict(meta.last_metadata.get("configuration", {})).get( "lakeapi.config", "{}" ) ) diff --git a/bmsdna/lakeapi/core/datasource.py b/bmsdna/lakeapi/core/datasource.py index d084f60..472a4f9 100644 --- a/bmsdna/lakeapi/core/datasource.py +++ b/bmsdna/lakeapi/core/datasource.py @@ -28,7 +28,7 @@ ) from bmsdna.lakeapi.core.log import get_logger from bmsdna.lakeapi.core.model import get_param_def -from bmsdna.lakeapi.core.types import DeltaOperatorTypes, OperatorType +from bmsdna.lakeapi.core.types import OperatorType from deltalake2db.delta_meta_retrieval import ( DataType, field_to_type, @@ -233,6 +233,7 @@ def get_df( self, filters: Optional[FilterType] = None, endpoint: endpoints = "request", + limit: int | None = None, ) -> ResultData: if self.df is None: unique_table_name = ( @@ -254,6 +255,7 @@ def get_df( self.config.file_type, filters=filters, meta_only=endpoint == "meta", + limit=limit, ) self.df = self.sql_context.execute_sql(self.query) @@ -264,7 +266,7 @@ def get_partition_filter( param: tuple[str, str], deltaMeta: DeltaMetadata, param_def: Iterable[Param | str], -) -> Optional[Tuple[str, OperatorType, Any]]: +) -> Optional[Tuple[str, SupportedOperator, Any]]: operators = ("<=", ">=", "=", "==", "in", "not in") key, value = param if not key or not value or key in ("limit", "offset"): @@ -274,7 +276,7 @@ def get_partition_filter( if prmdef_and_op is None: raise ValueError(f"thats not parameter: {key}") prmdef, op = prmdef_and_op - if op not in get_args(DeltaOperatorTypes): + if op not in get_args(SupportedOperator): return None colname = prmdef.colname or prmdef.name @@ -337,7 +339,7 @@ def get_partition_filter( # partition value must be string return ( col_for_partitioning, - op, + cast(SupportedOperator, op), value_for_partitioning, ) diff --git a/bmsdna/lakeapi/core/types.py b/bmsdna/lakeapi/core/types.py index 30bc4dd..7dc4f2a 100644 --- a/bmsdna/lakeapi/core/types.py +++ b/bmsdna/lakeapi/core/types.py @@ -54,7 +54,6 @@ "has", "startswith", ] -DeltaOperatorTypes = Literal["<", "=", ">", ">=", "<=", "in", "not in"] PolaryTypeFunction = Literal[ "sum", "count", diff --git a/bmsdna/lakeapi/endpoint/endpoint.py b/bmsdna/lakeapi/endpoint/endpoint.py index 2611567..ea66ed5 100644 --- a/bmsdna/lakeapi/endpoint/endpoint.py +++ b/bmsdna/lakeapi/endpoint/endpoint.py @@ -207,6 +207,8 @@ async def data( engine, chunk_size=real_chunk_size, ) + if not (limit == -1 and config.allow_get_all_pages): + limit = (1000 if limit == -1 else limit) or 1000 assert config.datasource is not None with Datasource( config.version_str, @@ -225,7 +227,20 @@ async def data( ) else: pre_filter = None - df = realdataframe.get_df(filters=pre_filter) + if config.datasource.file_type == "delta" and config.params is not None: + st = realdataframe.get_delta_table(True) + if st is not None and params is not None: + part_filter = filter_partitions_based_on_params( + st, params.model_dump(exclude_unset=True), config.params + ) + if part_filter: + pre_filter = list(pre_filter or []) + part_filter + df = realdataframe.get_df( + filters=pre_filter, + limit=limit + if not config.datasource.sortby and not offset and not limit == -1 + else None, # if sorted we need all data to sort correctly + ) df_cols = df.columns() expr = get_params_filter_expr( # this supports all kinds of filters, while the prefilter only supports equality context, diff --git a/pyproject.toml b/pyproject.toml index bb9357a..73e805b 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "bmsdna-lakeapi" -version = "0.25.1" +version = "0.25.2" description = "" authors = [{ name = "DWH Team", email = "you@example.com" }] dependencies = [ @@ -16,10 +16,9 @@ dependencies = [ "fastapi >=0.110.0", "arrow-odbc >=5.0.0", "expandvars >=0.12.0", - "pandas >=2.1.0,<3", "fsspec >=2024.2.0,<2025", "adlfs >=2024.2.0,<2025", - "deltalake2db >=1.1.1", + "deltalake2db >=1.1.4", ] requires-python = "~=3.10" @@ -32,6 +31,7 @@ polars = ["fastexcel"] auth = ["argon2-cffi", "pyjwt"] useradd = ["ruamel.yaml"] odbc = ["arrow-odbc"] +pandas = ["pandas >=2.1.0,<3"] [build-system] requires = ["hatchling"] diff --git a/tests/test_app.py b/tests/test_app.py index 3576f40..41f2d34 100644 --- a/tests/test_app.py +++ b/tests/test_app.py @@ -77,21 +77,20 @@ def test_fruits_sort_desc(client: TestClient): ] -def test_fruits_offset_1(client: TestClient): +@pytest.mark.parametrize("engine", engines) +def test_fruits_offset_1(engine, client: TestClient): for _ in range(2): - for e in engines: # polars does not support offset - response = client.get( - f"/api/v1/test/fruits?limit=1&&offset=1&format=json&%24engine={e}", - ) - assert response.status_code == 200 - assert response.json() == [ - { - "A": 2, - "fruits": "banana", - "B": 4, - "cars": "audi", - } - ] + response = client.get( + f"/api/v1/test/fruits?limit=1&&offset=0&format=json&%24engine={engine}", + ) + assert response.status_code == 200 + res = response.json() + response = client.get( + f"/api/v1/test/fruits?limit=1&&offset=1&format=json&%24engine={engine}", + ) + assert response.status_code == 200 + res2 = response.json() + assert res != res2 def test_data_limit(client: TestClient): diff --git a/uv.lock b/uv.lock index 21543b4..19d56e7 100644 --- a/uv.lock +++ b/uv.lock @@ -432,7 +432,7 @@ wheels = [ [[package]] name = "bmsdna-lakeapi" -version = "0.24.3" +version = "0.25.2" source = { editable = "." } dependencies = [ { name = "adlfs" }, @@ -444,7 +444,6 @@ dependencies = [ { name = "fastapi" }, { name = "fastexcel" }, { name = "fsspec" }, - { name = "pandas" }, { name = "polars" }, { name = "pyjwt" }, { name = "pyyaml" }, @@ -461,6 +460,9 @@ auth = [ odbc = [ { name = "arrow-odbc" }, ] +pandas = [ + { name = "pandas" }, +] polars = [ { name = "fastexcel" }, ] @@ -501,14 +503,14 @@ requires-dist = [ { name = "argon2-cffi", marker = "extra == 'auth'" }, { name = "arrow-odbc", specifier = ">=5.0.0" }, { name = "arrow-odbc", marker = "extra == 'odbc'" }, - { name = "deltalake2db", specifier = ">=1.1.1" }, + { name = "deltalake2db", specifier = ">=1.1.4" }, { name = "duckdb", specifier = ">=1.1.0,<2" }, { name = "expandvars", specifier = ">=0.12.0" }, { name = "fastapi", specifier = ">=0.110.0" }, { name = "fastexcel", specifier = ">=0.10.4" }, { name = "fastexcel", marker = "extra == 'polars'" }, { name = "fsspec", specifier = ">=2024.2.0,<2025" }, - { name = "pandas", specifier = ">=2.1.0,<3" }, + { name = "pandas", marker = "extra == 'pandas'", specifier = ">=2.1.0,<3" }, { name = "polars", specifier = ">=1.12.0,<2" }, { name = "pyjwt", specifier = ">=2.6.0,<3" }, { name = "pyjwt", marker = "extra == 'auth'" }, @@ -518,7 +520,7 @@ requires-dist = [ { name = "sqlglot", specifier = ">=24.0.0" }, { name = "xlsxwriter", specifier = ">=3.1.0,<4" }, ] -provides-extras = ["polars", "auth", "useradd", "odbc"] +provides-extras = ["polars", "auth", "useradd", "odbc", "pandas"] [package.metadata.requires-dev] dev = [ @@ -912,15 +914,15 @@ wheels = [ [[package]] name = "deltalake2db" -version = "1.1.1" +version = "1.1.4" source = { registry = "https://pypi.org/simple" } dependencies = [ { name = "polars" }, { name = "sqlglot" }, ] -sdist = { url = "https://files.pythonhosted.org/packages/8b/c3/0dac5043069d8b49ad846c97ba5ce3443eaf6bac248ec27f3e0a8bbdbbe9/deltalake2db-1.1.1.tar.gz", hash = "sha256:9aa1a62b3f1b0f2d79e8a028d0e43beb5bf3ea89ad599e14dbd622affe6b6994", size = 246492, upload-time = "2025-09-24T16:11:37.51Z" } +sdist = { url = "https://files.pythonhosted.org/packages/b3/7b/030086da48f107fcd7e39c94ebc22732ceaffefbab00a852e86fa2d0a2f2/deltalake2db-1.1.4.tar.gz", hash = "sha256:51de2b36f3a7c58dee9d7b7656537dff5433900231b7d13fa8c23c3e0275a1e7", size = 246904, upload-time = "2025-09-25T08:08:14.891Z" } wheels = [ - { url = "https://files.pythonhosted.org/packages/73/8a/a2da1c17994cd3d543303ea16aa3c3b33889f24b6856df0b3718cd1de0c4/deltalake2db-1.1.1-py3-none-any.whl", hash = "sha256:6f17caef7630d3b217d2898f30984755d824c6aaf44d6b7b0fd6daa9c91198a4", size = 19137, upload-time = "2025-09-24T16:11:36.128Z" }, + { url = "https://files.pythonhosted.org/packages/2a/ab/269c39628a2d2cf89357767ee32814de876052ed3c518e0fedd3d9217961/deltalake2db-1.1.4-py3-none-any.whl", hash = "sha256:2cba9bfabecc0664aa62856aea6882fc4279ccb9f048ebefa9071cda886b1dc9", size = 19540, upload-time = "2025-09-25T08:08:13.426Z" }, ] [[package]]