diff --git a/README.md b/README.md index 126f026..9755f50 100644 --- a/README.md +++ b/README.md @@ -5,7 +5,7 @@ [![CircleCI](https://circleci.com/gh/datagouv/api-tabular.svg?style=svg)](https://app.circleci.com/pipelines/github/datagouv/api-tabular) [![License: MIT](https://img.shields.io/badge/License-MIT-yellow.svg)](https://opensource.org/licenses/MIT) -An API service that provides RESTful access to CSV or tabular data converted by [Hydra](https://github.com/datagouv/hydra). This service provides a REST API to access PostgreSQL database tables containing CSV data, offering HTTP querying capabilities, pagination, and data streaming for CSV or tabular resources. +An API service that provides RESTful access to CSV or tabular data converted by [Hydra](https://github.com/datagouv/hydra). The **Tabular API** queries public Parquet files with [DuckDB](https://duckdb.org/) (filters, pagination, aggregations, CSV/JSON export), while resource metadata still comes from PostgreSQL via PostgREST (`tables_index`). The **Metrics API** continues to use PostgREST against PostgreSQL. This service is mainly used, developed and maintained by [data.gouv.fr](https://data.gouv.fr) - the France Open Data platform. The production API is deployed on data.gouv.fr infrastructure at [`https://tabular-api.data.gouv.fr/api`](https://tabular-api.data.gouv.fr/api). See the [product documentation](https://www.data.gouv.fr/dataservices/api-tabulaire-data-gouv-fr-beta/) (in French) for usage details and the [technical documentation](https://tabular-api.data.gouv.fr/api/doc) for API reference. @@ -74,9 +74,11 @@ The production API is deployed on data.gouv.fr infrastructure at [`https://tabul Query the API using a `resource_id`. Several test resources are available in the fake database: - **`aaaaaaaa-1111-bbbb-2222-cccccccccccc`** - Main test resource with 1000 rows - - **`aaaaaaaa-5555-bbbb-6666-cccccccccccc`** - Resource with database indexes - - **`dddddddd-7777-eeee-8888-ffffffffffff`** - Resource allowed for aggregation - - **`aaaaaaaa-9999-bbbb-1010-cccccccccccc`** - Resource with indexes and aggregation allowed + - **`aaaaaaaa-5555-bbbb-6666-cccccccccccc`** - Smaller sample resource + - **`dddddddd-7777-eeee-8888-ffffffffffff`** - Smaller sample resource + - **`aaaaaaaa-9999-bbbb-1010-cccccccccccc`** - Smaller sample resource + + Local Parquet fixtures live under `db/parquet/{resource_id}.parquet`. Set `PARQUET_BASE_URL` to that directory when running the Tabular API against the test stack (the pytest suite does this automatically). ### 🏭 Run with a real Hydra database @@ -320,9 +322,8 @@ column_name__sort=desc ``` #### Aggregation Operators -> ⚠️ **WARNING**: Aggregation requests are disabled by default. -> You can allow or disallow it for all resources with the `ALLOW_AGGREGATION` config. If `ALLOW_AGGREGATION` is set to `false`, you can specify allow exceptions by listing these in the `ALLOW_AGGREGATION_EXCEPTIONS` list. -> You can get the current status and exceptions at `/api/aggregation-exceptions/` endpoint. + +Aggregation is available on all resources and columns (except JSON limitations below). ``` # group by values @@ -486,16 +487,19 @@ Configuration is handled through TOML files and environment variables. The defau | Option | Default | Description | |--------|---------|-------------| -| `PGREST_ENDPOINT` | `http://localhost:8080` | PostgREST server URL | +| `PGREST_ENDPOINT` | `http://localhost:8080` | PostgREST server URL (metadata for Tabular API; data for Metrics API) | +| `S3_ENDPOINT` | `""` | S3 endpoint hosting public Parquet files (set via env in deploy) | +| `S3_BUCKET` | `""` | S3 bucket name (set via env in deploy) | +| `PARQUET_BASE_URL` | `""` | Optional override for Parquet base URL/path (tests use a local directory). When empty, URLs are `https://{S3_ENDPOINT}/{S3_BUCKET}/parquet/{resource_id}.parquet` | | `SERVER_NAME` | `localhost:8005` | Server name for URL generation | | `SCHEME` | `http` | URL scheme (http/https) | | `SENTRY_DSN` | `None` | Sentry DSN for error reporting (optional) | | `PAGE_SIZE_DEFAULT` | `20` | Default page size | | `PAGE_SIZE_MAX` | `50` | Maximum allowed page size | -| `BATCH_SIZE` | `50000` | Batch size for streaming | +| `BATCH_SIZE` | `50000` | Max rows for CSV/JSON export | | `DOC_PATH` | `/api/doc` | Swagger documentation path | -| `ALLOW_AGGREGATION` | `False` | Whether aggregation queries are allowed (`column__groupby`). If False, it can still be explicitly allowed with `ALLOW_AGGREGATION_EXCEPTIONS` | -| `ALLOW_AGGREGATION_EXCEPTIONS` | `["dddddddd-7777-eeee-8888-ffffffffffff", "aaaaaaaa-9999-bbbb-1010-cccccccccccc"]` | List of resource IDs allowed for aggregation | + +> Disk caching of Parquet files is not implemented yet; DuckDB reads remote (or local) Parquet via `httpfs` / filesystem. A local cache may be added later if latency requires it. ### Environment Variables @@ -503,13 +507,14 @@ You can override any configuration value using environment variables: ```shell export PGREST_ENDPOINT="http://my-postgrest:8080" +export S3_ENDPOINT="s3.example.com" +export S3_BUCKET="my-bucket" export PAGE_SIZE_DEFAULT=50 export SENTRY_DSN="https://your-sentry-dsn" ``` -Once the containers are up and running, you can directly query PostgREST on: + +PostgREST remains available for metadata inspection and for the Metrics API: `/?` -like for example: -`http://localhost:8080/eb7a008177131590c2f1a2ca0?decompte=eq.10` ### Custom Configuration File diff --git a/api_tabular/__init__.py b/api_tabular/__init__.py index 529be02..5df6faf 100644 --- a/api_tabular/__init__.py +++ b/api_tabular/__init__.py @@ -55,6 +55,21 @@ def check(self): """Sanity check on config""" pass + def parquet_url(self, resource_id: str) -> str: + """Public Parquet URL (or local path override) for a resource.""" + assert self.configuration is not None + base = (self.configuration.get("PARQUET_BASE_URL") or "").rstrip("/") + if not base: + endpoint = (self.configuration.get("S3_ENDPOINT") or "").strip() + bucket = (self.configuration.get("S3_BUCKET") or "").strip() + if not endpoint or not bucket: + raise ValueError( + "Parquet location is not configured: set PARQUET_BASE_URL, " + "or both S3_ENDPOINT and S3_BUCKET" + ) + base = f"https://{endpoint}/{bucket}/parquet" + return f"{base}/{resource_id}.parquet" + def __getattr__(self, __name): assert self.configuration is not None return self.configuration.get(__name) diff --git a/api_tabular/config_default.toml b/api_tabular/config_default.toml index cd3c418..73b5c54 100644 --- a/api_tabular/config_default.toml +++ b/api_tabular/config_default.toml @@ -1,4 +1,9 @@ PGREST_ENDPOINT = "http://localhost:8080" +# Set via environment (no defaults in repo). Used when PARQUET_BASE_URL is empty. +S3_ENDPOINT = "" +S3_BUCKET = "" +# Optional override for tests / local paths. When empty, URLs are built from S3_*. +PARQUET_BASE_URL = "" SERVER_NAME = "localhost:8005" SCHEME = "http" SENTRY_DSN = "" @@ -7,8 +12,3 @@ PAGE_SIZE_DEFAULT = 20 PAGE_SIZE_MAX = 50 BATCH_SIZE = 50000 DOC_PATH = "/api/doc" -ALLOW_AGGREGATION = false -ALLOW_AGGREGATION_EXCEPTIONS = [ - "dddddddd-7777-eeee-8888-ffffffffffff", # without indexes - "aaaaaaaa-9999-bbbb-1010-cccccccccccc", # with indexes -] # list of resource_ids diff --git a/api_tabular/core/query.py b/api_tabular/core/query.py index 2cab7d1..8778a61 100644 --- a/api_tabular/core/query.py +++ b/api_tabular/core/query.py @@ -1,8 +1,6 @@ import re from collections import defaultdict -from api_tabular.core.utils import is_aggregation_allowed - def build_sql_query_string( request_arg: list, @@ -11,6 +9,7 @@ def build_sql_query_string( page_size: int | None = None, offset: int = 0, ) -> str: + """Build a PostgREST query string (used by the Metrics API).""" sql_query = [] aggregators = defaultdict(list) sorted = False @@ -39,11 +38,6 @@ def build_sql_query_string( else: raise ValueError(f"argument '{arg}' could not be parsed") if aggregators: - if resource_id and not is_aggregation_allowed(resource_id): - raise PermissionError( - f"Aggregation parameters `{'`, `'.join(aggregators.keys())}` " - f"are not allowed for resource '{resource_id}'" - ) agg_query = "select=" for operator in aggregators: if operator == "groupby": diff --git a/api_tabular/core/swagger.py b/api_tabular/core/swagger.py index 5415ee4..155e0cc 100644 --- a/api_tabular/core/swagger.py +++ b/api_tabular/core/swagger.py @@ -2,8 +2,6 @@ import yaml -from api_tabular.core.utils import is_aggregation_allowed - TYPE_POSSIBILITIES = { "string": [ "isnull", @@ -200,10 +198,6 @@ def swagger_parameters(resource_columns: dict, resource_id: str) -> list: # see cast for db here: https://github.com/datagouv/csv-detective/blob/master/csv_detective/output/dataframe.py for key, value in resource_columns.items(): for op in OPERATORS_DESCRIPTIONS: - if not is_aggregation_allowed(resource_id) and OPERATORS_DESCRIPTIONS[op].get( - "is_aggregator" - ): - continue if op in TYPE_POSSIBILITIES[value["python_type"]]: op_name = cast(str, OPERATORS_DESCRIPTIONS[op]["name"]) op_description = cast(str, OPERATORS_DESCRIPTIONS[op]["description"]) diff --git a/api_tabular/core/utils.py b/api_tabular/core/utils.py index 1aba153..434cff2 100644 --- a/api_tabular/core/utils.py +++ b/api_tabular/core/utils.py @@ -5,10 +5,6 @@ from api_tabular.core.error import QueryException -def is_aggregation_allowed(resource_id: str) -> bool: - return config.ALLOW_AGGREGATION or resource_id in config.ALLOW_AGGREGATION_EXCEPTIONS - - def process_total(res: Response | ClientResponse) -> int: # the Content-Range looks like this: '0-49/21777' # see https://docs.postgrest.org/en/stable/references/api/pagination_count.html diff --git a/api_tabular/tabular/app.py b/api_tabular/tabular/app.py index 4feb496..113e0cc 100644 --- a/api_tabular/tabular/app.py +++ b/api_tabular/tabular/app.py @@ -15,7 +15,6 @@ from api_tabular.core.utils import build_offset from api_tabular.core.version import get_app_version from api_tabular.tabular.utils import ( - get_potential_indexes, get_resource, get_resource_data, stream_resource_data, @@ -62,8 +61,6 @@ async def resource_profile(request): resource: dict = await get_resource( request.app["csession"], resource_id, ["profile:csv_detective"] ) - indexes: set | None = await get_potential_indexes(request.app["csession"], resource_id) - resource["indexes"] = list(indexes) if isinstance(indexes, set) else None return web.json_response(resource) @@ -73,10 +70,7 @@ async def resource_swagger(request): resource: dict = await get_resource( request.app["csession"], resource_id, ["profile:csv_detective"] ) - indexes: set | None = await get_potential_indexes(request.app["csession"], resource_id) columns: dict[str, str] = resource["profile"]["columns"] - if indexes: - columns = {col: params for col, params in columns.items() if col in indexes} swagger_string = build_swagger_file(columns, resource_id) return web.Response(body=swagger_string) @@ -102,9 +96,9 @@ async def resource_data(request): offset = build_offset(page, page_size) - sql_query = await try_build_query(request, query_string, resource_id, page_size, offset) - resource = await get_resource(request.app["csession"], resource_id, ["parsing_table"]) - response, total = await get_resource_data(request.app["csession"], resource, sql_query) + query = await try_build_query(request, query_string, resource_id, page_size, offset) + await get_resource(request.app["csession"], resource_id, []) + response, total = await get_resource_data(resource_id, query) next = build_link_with_page(request, query_string, page + 1, page_size) prev = build_link_with_page(request, query_string, page - 1, page_size) @@ -146,16 +140,6 @@ async def get_health(request): return await check_health(request, f"{config.PGREST_ENDPOINT}/migrations_csv") -@routes.get(r"/api/aggregation-exceptions/") -async def get_aggregation_exceptions(request): - """Return the list of resources for which aggregation queries are allowed""" - body = { - "allowed": config.ALLOW_AGGREGATION, - "exceptions": config.ALLOW_AGGREGATION_EXCEPTIONS, - } - return web.json_response(body) - - async def app_factory(): async def on_startup(app): app["csession"] = ClientSession() diff --git a/api_tabular/tabular/duckdb_exec.py b/api_tabular/tabular/duckdb_exec.py new file mode 100644 index 0000000..9c82764 --- /dev/null +++ b/api_tabular/tabular/duckdb_exec.py @@ -0,0 +1,208 @@ +"""Execute DuckDB queries against remote or local Parquet files.""" + +from __future__ import annotations + +import asyncio +import csv +import json +from io import StringIO +from typing import Any + +import duckdb +from aiohttp import web +from aiohttp.web import StreamResponse +from aiohttp.web_request import Request + +from api_tabular import config +from api_tabular.core.error import QueryException +from api_tabular.tabular.duckdb_query import DuckDBQuery, quote_ident + + +def _connect() -> duckdb.DuckDBPyConnection: + con = duckdb.connect(database=":memory:") + # httpfs is needed for https:// Parquet URLs; harmless for local files + try: + con.execute("INSTALL httpfs;") + except duckdb.Error: + pass + try: + con.execute("LOAD httpfs;") + except duckdb.Error: + pass + return con + + +def _parquet_source_sql(parquet_path: str, columns: list[str]) -> tuple[str, list]: + """Build FROM clause ensuring a synthetic __id when missing from the file.""" + if "__id" in columns: + return "read_parquet(?)", [parquet_path] + return ( + f"(SELECT file_row_number AS {quote_ident('__id')}, " + f"* EXCLUDE (file_row_number) FROM read_parquet(?, file_row_number = true))", + [parquet_path], + ) + + +def _describe_columns(con: duckdb.DuckDBPyConnection, parquet_path: str) -> list[str]: + rows = con.execute("DESCRIBE SELECT * FROM read_parquet(?)", [parquet_path]).fetchall() + return [row[0] for row in rows] + + +def _assemble_sql( + source_sql: str, + query: DuckDBQuery, + *, + for_count: bool = False, +) -> str: + if for_count: + return f"SELECT COUNT(*) FROM {source_sql} AS src {query.where_sql}" + + parts = [ + f"SELECT {query.select_sql} FROM {source_sql} AS src", + query.where_sql, + query.group_by_sql, + query.order_sql, + ] + sql = " ".join(p for p in parts if p) + if query.limit is not None: + sql += f" LIMIT {int(query.limit)}" + if query.offset >= 1: + sql += f" OFFSET {int(query.offset)}" + return sql + + +def _json_safe(value: Any) -> Any: + if value is None or isinstance(value, (str, int, float, bool)): + return value + if isinstance(value, (list, dict)): + return value + return str(value) + + +def _rows_to_dicts(con: duckdb.DuckDBPyConnection, result) -> list[dict[str, Any]]: + columns = [desc[0] for desc in result.description] + return [{col: _json_safe(val) for col, val in zip(columns, row)} for row in result.fetchall()] + + +def _execute_sync( + parquet_path: str, + query: DuckDBQuery, +) -> tuple[list[dict[str, Any]], int | None]: + con = _connect() + try: + try: + columns = _describe_columns(con, parquet_path) + except duckdb.Error as e: + message = str(e).lower() + if "404" in message or "not found" in message or "no files found" in message: + raise web.HTTPNotFound() from e + raise QueryException(400, None, "Database error", str(e)) from e + + source_sql, source_params = _parquet_source_sql(parquet_path, columns) + data_sql = _assemble_sql(source_sql, query, for_count=False) + data_params = source_params + query.where_params + + try: + result = con.execute(data_sql, data_params) + rows = _rows_to_dicts(con, result) + except duckdb.Error as e: + raise QueryException(400, None, "Database error", str(e)) from e + + total: int | None = None + if not query.is_aggregation: + count_sql = _assemble_sql(source_sql, query, for_count=True) + count_params = source_params + query.where_params + total = int( + con.execute(count_sql, count_params).fetchone()[0] + ) # ty: ignore[not-subscriptable] + + return rows, total + finally: + con.close() + + +def _count_sync(parquet_path: str, query: DuckDBQuery) -> int: + con = _connect() + try: + try: + columns = _describe_columns(con, parquet_path) + except duckdb.Error as e: + message = str(e).lower() + if "404" in message or "not found" in message or "no files found" in message: + raise web.HTTPNotFound() from e + raise QueryException(400, None, "Database error", str(e)) from e + + source_sql, source_params = _parquet_source_sql(parquet_path, columns) + count_sql = _assemble_sql(source_sql, query, for_count=True) + return int( + con.execute(count_sql, source_params + query.where_params).fetchone()[ + 0 + ] # ty: ignore[not-subscriptable] + ) + except duckdb.Error as e: + raise QueryException(400, None, "Database error", str(e)) from e + finally: + con.close() + + +def _fetch_all_sync(parquet_path: str, query: DuckDBQuery) -> list[dict[str, Any]]: + rows, _ = _execute_sync(parquet_path, query) + return rows + + +async def execute_query( + resource_id: str, + query: DuckDBQuery, +) -> tuple[list[dict[str, Any]], int | None]: + parquet_path = config.parquet_url(resource_id) + return await asyncio.to_thread(_execute_sync, parquet_path, query) + + +async def count_rows(resource_id: str, query: DuckDBQuery) -> int: + parquet_path = config.parquet_url(resource_id) + return await asyncio.to_thread(_count_sync, parquet_path, query) + + +async def stream_parquet_data( + request: Request, + resource_id: str, + query: DuckDBQuery, + format: str, + response_headers: dict, +) -> StreamResponse: + total = await count_rows(resource_id, query) + if total > config.BATCH_SIZE: + raise QueryException( + 403, + None, + "Output is too long", + f"The output has more than {config.BATCH_SIZE} rows, please consider downloading the source file directly", + ) + + export_query = DuckDBQuery( + where_sql=query.where_sql, + where_params=list(query.where_params), + order_sql=query.order_sql, + select_sql=query.select_sql, + group_by_sql=query.group_by_sql, + is_aggregation=query.is_aggregation, + limit=config.BATCH_SIZE, + offset=0, + ) + rows = await asyncio.to_thread(_fetch_all_sync, config.parquet_url(resource_id), export_query) + + response = web.StreamResponse(headers=response_headers) + await response.prepare(request) + + if format == "json": + await response.write(json.dumps(rows, default=str).encode()) + else: + buffer = StringIO() + if rows: + writer = csv.DictWriter(buffer, fieldnames=list(rows[0].keys())) + writer.writeheader() + writer.writerows(rows) + await response.write(buffer.getvalue().encode()) + + await response.write_eof() + return response diff --git a/api_tabular/tabular/duckdb_query.py b/api_tabular/tabular/duckdb_query.py new file mode 100644 index 0000000..3d4bb44 --- /dev/null +++ b/api_tabular/tabular/duckdb_query.py @@ -0,0 +1,292 @@ +import re +from collections import defaultdict +from dataclasses import dataclass, field +from typing import Any + + +@dataclass +class DuckDBQuery: + """SQL fragments ready to run against a Parquet source aliased as `src`.""" + + where_sql: str = "" + where_params: list = field(default_factory=list) + order_sql: str = "" + select_sql: str = "*" + group_by_sql: str = "" + is_aggregation: bool = False + limit: int | None = None + offset: int = 0 + + +@dataclass +class FilterAction: + kind: str # where | order | select + sql: str = "" + params: list[Any] = field(default_factory=list) + columns: list[str] = field(default_factory=list) + + +def quote_ident(name: str) -> str: + return '"' + name.replace('"', '""') + '"' + + +def get_column_and_operator(argument: str) -> tuple[str, str]: + *column_split, comparator = argument.split("__") + column = "__".join(column_split) + return column, comparator.lower() + + +def build_duckdb_query( + request_arg: list[str], + page_size: int | None = None, + offset: int = 0, +) -> DuckDBQuery: + where_parts: list[str] = [] + where_params: list[Any] = [] + order_sql = "" + select_columns: list[str] | None = None + aggregators: dict[str, list[str]] = defaultdict(list) + has_sort = False + + for arg in request_arg: + if arg.startswith("or=("): + clause, params = parse_operator(query=arg, operator="or", top_level=True) + where_parts.append(clause) + where_params.extend(params) + continue + _split = arg.split("=") + if len(_split) == 2: + action = add_filter(*_split) + if action is None: + continue + if action.kind == "where": + where_parts.append(action.sql) + where_params.extend(action.params) + elif action.kind == "order": + order_sql = action.sql + has_sort = True + elif action.kind == "select": + select_columns = action.columns + elif len(_split) == 1: + if "__" not in _split[0]: + raise ValueError(f"argument '{arg}' could not be parsed") + if _split[0].split("__")[-1] in ["isnull", "isnotnull"]: + action = add_filter(_split[0], "") + assert action is not None and action.kind == "where" + where_parts.append(action.sql) + where_params.extend(action.params) + else: + column, operator = add_aggregator(_split[0]) + aggregators[operator].append(column) + else: + raise ValueError(f"argument '{arg}' could not be parsed") + + is_aggregation = bool(aggregators) + if is_aggregation and select_columns is not None: + raise ValueError("the argument `columns` cannot be set alongside aggregators") + + select_sql = "*" + group_by_sql = "" + if is_aggregation: + select_parts: list[str] = [] + group_cols: list[str] = [] + for operator, columns in aggregators.items(): + if operator == "groupby": + for column in columns: + ident = quote_ident(column) + select_parts.append(ident) + group_cols.append(ident) + else: + for column in columns: + ident = quote_ident(column) + alias = quote_ident(f"{column}__{operator}") + select_parts.append(f"{operator.upper()}({ident}) AS {alias}") + select_sql = ", ".join(select_parts) + if group_cols: + group_by_sql = ", ".join(group_cols) + elif select_columns is not None: + select_sql = ", ".join(quote_ident(c) for c in select_columns) + + if not has_sort and not is_aggregation: + order_sql = f"ORDER BY {quote_ident('__id')} ASC" + + where_sql = "" + if where_parts: + where_sql = "WHERE " + " AND ".join(where_parts) + + return DuckDBQuery( + where_sql=where_sql, + where_params=where_params, + order_sql=order_sql, + select_sql=select_sql, + group_by_sql=f"GROUP BY {group_by_sql}" if group_by_sql else "", + is_aggregation=is_aggregation, + limit=page_size, + offset=offset, + ) + + +def add_filter( + argument: str, + value: str, + *, + in_operator: bool = False, +) -> FilterAction | None: + if argument in ["page", "page_size"]: + if in_operator: + raise ValueError(f"Argument `{argument}` can't be set within an operator") + return None + if argument == "columns": + if in_operator: + raise ValueError(f"Argument `{argument}` can't be set within an operator") + if '"' in value: + raise ValueError('Forbidden character `"` in a column name') + return FilterAction(kind="select", columns=value.split(",")) + if "__" not in argument: + raise ValueError(f"argument '{argument}={value}' could not be parsed") + + column, normalized_comparator = get_column_and_operator(argument) + ident = quote_ident(column) + + if normalized_comparator == "sort": + if in_operator: + raise ValueError(f"Argument `{argument}` can't be set within an operator") + direction = value.upper() + if direction not in ("ASC", "DESC"): + raise ValueError(f"argument '{argument}={value}' could not be parsed") + return FilterAction(kind="order", sql=f"ORDER BY {ident} {direction}") + if normalized_comparator == "exact": + return FilterAction(kind="where", sql=f"{ident} = ?", params=[coerce_value(value)]) + if normalized_comparator == "differs": + return FilterAction( + kind="where", sql=f"{ident} IS DISTINCT FROM ?", params=[coerce_value(value)] + ) + if normalized_comparator == "isnull": + return FilterAction(kind="where", sql=f"{ident} IS NULL") + if normalized_comparator == "isnotnull": + return FilterAction(kind="where", sql=f"{ident} IS NOT NULL") + if normalized_comparator == "contains": + return FilterAction(kind="where", sql=f"{ident} ILIKE '%' || ? || '%'", params=[value]) + if normalized_comparator == "notcontains": + return FilterAction( + kind="where", sql=f"NOT ({ident} ILIKE '%' || ? || '%')", params=[value] + ) + if normalized_comparator == "in": + values = value.split(",") + placeholders = ", ".join("?" for _ in values) + return FilterAction( + kind="where", + sql=f"{ident} IN ({placeholders})", + params=[coerce_value(v) for v in values], + ) + if normalized_comparator == "notin": + values = value.split(",") + placeholders = ", ".join("?" for _ in values) + return FilterAction( + kind="where", + sql=f"{ident} NOT IN ({placeholders})", + params=[coerce_value(v) for v in values], + ) + if normalized_comparator == "less": + return FilterAction(kind="where", sql=f"{ident} <= ?", params=[coerce_value(value)]) + if normalized_comparator == "greater": + return FilterAction(kind="where", sql=f"{ident} >= ?", params=[coerce_value(value)]) + if normalized_comparator == "strictly_less": + return FilterAction(kind="where", sql=f"{ident} < ?", params=[coerce_value(value)]) + if normalized_comparator == "strictly_greater": + return FilterAction(kind="where", sql=f"{ident} > ?", params=[coerce_value(value)]) + raise ValueError(f"argument '{argument}={value}' could not be parsed") + + +def coerce_value(value: str): + """Coerce query-string values. Keep numbers as strings so DuckDB can cast to the column type.""" + lower = value.lower() + if lower == "true": + return True + if lower == "false": + return False + return value + + +def add_aggregator(argument: str) -> tuple[str, str]: + if "__" not in argument: + raise ValueError(f"argument '{argument}' could not be parsed") + column, operator = get_column_and_operator(argument) + if operator in ["avg", "count", "max", "min", "sum", "groupby"]: + return column, operator + raise ValueError(f"argument '{argument}' could not be parsed") + + +def split_top_level(s: str) -> list[str]: + parts = [] + current = "" + depth = 0 + for char in s: + if char == "(": + depth += 1 + current += char + elif char == ")": + depth -= 1 + current += char + elif char == "," and depth == 0: + parts.append(current) + current = "" + else: + current += char + if current: + parts.append(current) + return parts + + +def find_arg_val(param: str) -> tuple[str, str]: + if param.count('"') not in {0, 2, 4}: + raise ValueError(f"argument '{param}' could not be parsed") + column_operator_pattern = r'^"[^"]*"__[a-z]+' + value_pattern = r'\."[^"]*"$' + if param.count('"') == 0: + # col__op.val — value may contain dots (e.g. 0.1) + if "." not in param: + raise ValueError(f"argument '{param}' could not be parsed") + argument, value = param.split(".", 1) + if "__" not in argument: + raise ValueError(f"argument '{param}' could not be parsed") + return argument, value + if param.count('"') == 4: + col_op = re.findall(column_operator_pattern, param) + val = re.findall(value_pattern, param) + if len(col_op) != 1 or len(val) != 1: + raise ValueError(f"argument '{param}' could not be parsed") + return col_op[0].replace('"', ""), val[0][1:] + col_op = re.findall(column_operator_pattern, param) + val = re.findall(value_pattern, param) + if not col_op: + return param.split(".", 1)[0], val[0][1:] + return col_op[0].replace('"', ""), param.split(".")[-1] + + +def parse_operator(query: str, operator: str, top_level: bool = False) -> tuple[str, list]: + if not query.endswith(")"): + raise ValueError(f"argument '{query}' could not be parsed") + params_sql: list[str] = [] + params_values: list = [] + inner = re.findall(rf"^{operator}{'=' if top_level else ''}\((.*)\)$", query)[0] + for param in split_top_level(inner): + if param.startswith(("and(", "or(")): + clause, values = parse_operator(query=param, operator=param.split("(")[0]) + params_sql.append(clause) + params_values.extend(values) + elif param.endswith(("__isnull", "__isnotnull")): + action = add_filter(param.replace('"', ""), "", in_operator=True) + assert action is not None and action.kind == "where" + params_sql.append(action.sql) + params_values.extend(action.params) + else: + argument, value = find_arg_val(param) + if len(value) >= 2 and value.startswith('"') and value.endswith('"'): + value = value[1:-1] + action = add_filter(argument, value, in_operator=True) + assert action is not None and action.kind == "where" + params_sql.append(action.sql) + params_values.extend(action.params) + joiner = f" {operator.upper()} " + return f"({joiner.join(params_sql)})", params_values diff --git a/api_tabular/tabular/utils.py b/api_tabular/tabular/utils.py index 999c8ed..8e172a6 100644 --- a/api_tabular/tabular/utils.py +++ b/api_tabular/tabular/utils.py @@ -2,10 +2,9 @@ from aiohttp.web_request import Request from api_tabular import config -from api_tabular.core.data import stream_data from api_tabular.core.error import QueryException, handle_exception -from api_tabular.core.query import build_sql_query_string -from api_tabular.core.utils import process_total +from api_tabular.tabular.duckdb_exec import execute_query, stream_parquet_data +from api_tabular.tabular.duckdb_query import build_duckdb_query async def get_resource(session: ClientSession, resource_id: str, columns: list) -> dict: @@ -34,35 +33,8 @@ async def get_resource(session: ClientSession, resource_id: str, columns: list) return record[0] -async def get_resource_data( - session: ClientSession, resource: dict, sql_query: str -) -> tuple[list[dict], int | None]: - headers = {"Prefer": "count=exact"} - url = f"{config.PGREST_ENDPOINT}/{resource['parsing_table']}?{sql_query}" - skip_total = False - if any(f".{agg}()" in url for agg in ["count", "max", "min", "sum", "avg"]): - # the total for aggretated data is wrong, it is always the length of the original table - skip_total = True - async with session.get(url, headers=headers) as res: - if not res.ok: - handle_exception(res.status, "Database error", await res.json(), resource.get("id")) - record = await res.json() - total = process_total(res) if not skip_total else None - return record, total - - -async def get_potential_indexes(session: ClientSession, resource_id: str) -> set[str] | None: - q = f"select=table_indexes&resource_id=eq.{resource_id}" - url = f"{config.PGREST_ENDPOINT}/resources_exceptions?{q}" - async with session.get(url) as res: - record = await res.json() - if not res.ok: - handle_exception(res.status, "Database error", record, resource_id) - if not record: - return None - # indexes look like {"column_name": "index_type", ...} or None - indexes: dict = record[0].get("table_indexes", {}) - return set(indexes.keys()) if indexes else None +async def get_resource_data(resource_id: str, query) -> tuple[list[dict], int | None]: + return await execute_query(resource_id, query) async def try_build_query( @@ -72,22 +44,18 @@ async def try_build_query( page_size: int | None = None, offset: int = 0, ): - indexes: set | None = await get_potential_indexes(request.app["csession"], resource_id) try: - sql_query = build_sql_query_string(query_string, resource_id, indexes, page_size, offset) + return build_duckdb_query(query_string, page_size=page_size, offset=offset) except ValueError as e: raise QueryException(400, None, "Invalid query string", f"Malformed query: {e}") - except PermissionError as e: - raise QueryException(403, None, "Unauthorized parameters", str(e)) - return sql_query async def stream_resource_data(request: Request, format: str): resource_id = request.match_info["rid"] query_string = request.query_string.split("&") if request.query_string else [] - sql_query = await try_build_query(request, query_string, resource_id) - resource = await get_resource(request.app["csession"], resource_id, ["parsing_table"]) + query = await try_build_query(request, query_string, resource_id) + await get_resource(request.app["csession"], resource_id, []) mime = "application/json" if format == "json" else "text/csv" response_headers = { @@ -95,10 +63,10 @@ async def stream_resource_data(request: Request, format: str): "Content-Type": mime, } - return await stream_data( - session=request.app["csession"], + return await stream_parquet_data( request=request, - url=f"{config.PGREST_ENDPOINT}/{resource['parsing_table']}?{sql_query}", - accept_format=mime, + resource_id=resource_id, + query=query, + format=format, response_headers=response_headers, ) diff --git a/db/parquet/aaaaaaaa-1111-bbbb-2222-cccccccccccc.parquet b/db/parquet/aaaaaaaa-1111-bbbb-2222-cccccccccccc.parquet new file mode 100644 index 0000000..267d635 Binary files /dev/null and b/db/parquet/aaaaaaaa-1111-bbbb-2222-cccccccccccc.parquet differ diff --git a/db/parquet/aaaaaaaa-5555-bbbb-6666-cccccccccccc.parquet b/db/parquet/aaaaaaaa-5555-bbbb-6666-cccccccccccc.parquet new file mode 100644 index 0000000..a4ad0ce Binary files /dev/null and b/db/parquet/aaaaaaaa-5555-bbbb-6666-cccccccccccc.parquet differ diff --git a/db/parquet/aaaaaaaa-9999-bbbb-1010-cccccccccccc.parquet b/db/parquet/aaaaaaaa-9999-bbbb-1010-cccccccccccc.parquet new file mode 100644 index 0000000..273843d Binary files /dev/null and b/db/parquet/aaaaaaaa-9999-bbbb-1010-cccccccccccc.parquet differ diff --git a/db/parquet/dddddddd-1111-eeee-1212-ffffffffffff.parquet b/db/parquet/dddddddd-1111-eeee-1212-ffffffffffff.parquet new file mode 100644 index 0000000..784424c Binary files /dev/null and b/db/parquet/dddddddd-1111-eeee-1212-ffffffffffff.parquet differ diff --git a/db/parquet/dddddddd-3333-eeee-4444-ffffffffffff.parquet b/db/parquet/dddddddd-3333-eeee-4444-ffffffffffff.parquet new file mode 100644 index 0000000..485cc88 Binary files /dev/null and b/db/parquet/dddddddd-3333-eeee-4444-ffffffffffff.parquet differ diff --git a/db/parquet/dddddddd-7777-eeee-8888-ffffffffffff.parquet b/db/parquet/dddddddd-7777-eeee-8888-ffffffffffff.parquet new file mode 100644 index 0000000..273843d Binary files /dev/null and b/db/parquet/dddddddd-7777-eeee-8888-ffffffffffff.parquet differ diff --git a/docker-compose.yml b/docker-compose.yml index 469a9b8..7387550 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -59,6 +59,11 @@ services: - "8005:8005" environment: - PGREST_ENDPOINT=http://postgrest-test:8080 + - PARQUET_BASE_URL=/home/datagouv/db/parquet + - S3_ENDPOINT=${S3_ENDPOINT:-} + - S3_BUCKET=${S3_BUCKET:-} + volumes: + - ./db/parquet:/home/datagouv/db/parquet:ro depends_on: - postgrest-test profiles: diff --git a/pyproject.toml b/pyproject.toml index 18d9ded..e65a760 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -12,6 +12,7 @@ authors = [ dependencies = [ "aiohttp>=3.14.1,<4.0.0", "aiohttp-swagger>=1.0.16,<2.0.0", + "duckdb>=1.5.4", "gunicorn<24.0.0,>=23.0.0", "sentry-sdk>=2.49.0,<3.0.0", ] diff --git a/scripts/generate_test_parquets.py b/scripts/generate_test_parquets.py new file mode 100644 index 0000000..4e3273d --- /dev/null +++ b/scripts/generate_test_parquets.py @@ -0,0 +1,119 @@ +#!/usr/bin/env python3 +"""Regenerate db/parquet/*.parquet fixtures from db/initdb/0-init.sql COPY blocks.""" + +from __future__ import annotations + +import csv +import re +from pathlib import Path + +import duckdb + +ROOT = Path(__file__).resolve().parent.parent + + +def parse_cols(cols_raw: str) -> list[str]: + cols: list[str] = [] + i = 0 + s = cols_raw.strip() + while i < len(s): + if s[i] in " \t\n,": + i += 1 + continue + if s[i] == '"': + i += 1 + buf: list[str] = [] + while i < len(s): + if s[i] == '"' and i + 1 < len(s) and s[i + 1] == '"': + buf.append('"') + i += 2 + elif s[i] == '"': + i += 1 + break + else: + buf.append(s[i]) + i += 1 + cols.append("".join(buf)) + else: + m = re.match(r"[^\s,]+", s[i:]) + assert m is not None + cols.append(m.group(0)) + i += len(m.group(0)) + return cols + + +def qident(name: str) -> str: + return '"' + name.replace('"', '""') + '"' + + +def main() -> None: + sql = (ROOT / "db/initdb/0-init.sql").read_text() + tables_index = { + row["parsing_table"]: row["resource_id"] + for row in csv.DictReader(open(ROOT / "db/tables_index.csv")) + } + out_dir = ROOT / "db" / "parquet" + out_dir.mkdir(exist_ok=True) + for path in out_dir.glob("*.parquet"): + path.unlink() + + con = duckdb.connect() + for m in re.finditer(r'COPY "csvapi"\.(\w+) \((.*)\) FROM stdin;\n', sql): + table = m.group(1) + cols_raw = m.group(2) + start = m.end() + end = sql.find("\\.\n", start) + data = sql[start:end] + resource_id = tables_index.get(table) + if not resource_id: + continue + cols = parse_cols(cols_raw) + rows = [] + for line in data.split("\n"): + if not line.strip(): + continue + values = line.split("\t") + if len(values) != len(cols): + raise ValueError(f"{table}: column/value mismatch") + rows.append({c: (None if v == "\\N" else v) for c, v in zip(cols, values)}) + + path = out_dir / f"{resource_id}.parquet" + con.execute("DROP TABLE IF EXISTS tmp_export") + col_defs = ", ".join(f"{qident(c)} VARCHAR" for c in cols) + con.execute(f"CREATE TABLE tmp_export ({col_defs})") + placeholders = ", ".join(["?"] * len(cols)) + quoted_cols = ", ".join(qident(c) for c in cols) + for row in rows: + con.execute( + f"INSERT INTO tmp_export ({quoted_cols}) VALUES ({placeholders})", + [row[c] for c in cols], + ) + + select_parts = [] + for c in cols: + q = qident(c) + if c == "__id": + select_parts.append(f"CAST({q} AS INTEGER) AS {q}") + elif c == "score": + select_parts.append(f"TRY_CAST({q} AS DOUBLE) AS {q}") + elif c == "decompte": + select_parts.append(f"TRY_CAST({q} AS INTEGER) AS {q}") + elif c == "is_true": + select_parts.append( + f"CASE WHEN lower(cast({q} AS VARCHAR)) IN ('t','true','1') THEN true " + f"WHEN lower(cast({q} AS VARCHAR)) IN ('f','false','0') THEN false " + f"ELSE NULL END AS {q}" + ) + else: + select_parts.append(q) + con.execute( + f"CREATE OR REPLACE TABLE tmp_export AS SELECT {', '.join(select_parts)} FROM tmp_export" + ) + con.execute(f"COPY tmp_export TO '{path}' (FORMAT PARQUET, COMPRESSION ZSTD)") + count_row = con.execute("SELECT COUNT(*) FROM tmp_export").fetchone() + n = 0 if count_row is None else count_row[0] + print(f"wrote {path.name} ({n} rows)") + + +if __name__ == "__main__": + main() diff --git a/tabular_swagger.yaml b/tabular_swagger.yaml index fda2cf2..546f780 100644 --- a/tabular_swagger.yaml +++ b/tabular_swagger.yaml @@ -150,23 +150,6 @@ paths: type: string format: uuid example: "aaaaaaaa-1111-bbbb-2222-cccccccccccc" - /api/aggregation-exceptions/: - get: - tags: - - Data retrieval - description: Returns the list of resource IDs for which aggregation queries (groupby, count, sum, etc.) are allowed. - summary: List aggregation-allowed resources - operationId: getAggregationExceptions - responses: - '200': - description: List of resource UUIDs - content: - application/json: - schema: - type: array - items: - type: string - format: uuid /health/: get: tags: @@ -186,17 +169,11 @@ paths: components: schemas: ResourceProfileResponse: - description: Profile endpoint response; contains the csv_detective profile and optional column indexes. + description: Profile endpoint response; contains the csv_detective profile. type: object properties: profile: $ref: '#/components/schemas/ResourceProfile' - indexes: - description: List of column names that have an index (for filtering/sorting). Null if none. - type: array - items: - type: string - nullable: true ResourceProfile: description: Column types, formats and statistics from csv_detective (encoding, separator, row count, etc.) type: object diff --git a/tests/conftest.py b/tests/conftest.py index ab13050..d8df34b 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -1,10 +1,12 @@ import csv +import json from datetime import datetime, timezone from pathlib import Path from typing import Any, AsyncGenerator, Generator import pytest import pytest_asyncio +from aiohttp import web from aiohttp.test_utils import TestClient, TestServer from aioresponses import aioresponses @@ -19,24 +21,15 @@ DELETED_RESOURCE_ID = "deadbeef-dead-beef-dead-beefdeadbeef" NULL_VALUES_RESOURCE_ID = "dddddddd-1111-eeee-1212-ffffffffffff" +PARQUET_DIR = Path(__file__).parent.parent / "db" / "parquet" -@pytest.fixture -def rmock(): - # passthrough for local requests (aiohttp TestServer) - with aioresponses(passthrough=["http://127.0.0.1"]) as m: - yield m - -@pytest_asyncio.fixture -async def client() -> AsyncGenerator[TestClient, Any]: - app = await app_factory() - async with TestClient(TestServer(app)) as client: - yield client - - -@pytest.fixture -def base_url() -> Generator[str, Any, Any]: - yield f"{config.SCHEME}://{config.SERVER_NAME}" +@pytest.fixture(autouse=True) +def parquet_base_url(): + """Point DuckDB at local fixture Parquet files instead of S3.""" + config.override(PARQUET_BASE_URL=str(PARQUET_DIR.resolve())) + yield + config.override(PARQUET_BASE_URL="") def timestamptz_to_utc_iso(date_str: str) -> str: @@ -75,3 +68,65 @@ def tables_index_rows() -> Generator[dict, Any, Any]: @pytest.fixture def exceptions_rows() -> Generator[dict, Any, Any]: yield csv_to_dict("exceptions") + + +@pytest.fixture(autouse=True) +def mock_tables_index(mocker, tables_index_rows): + """Serve `tables_index` metadata from CSV fixtures (no PostgREST required).""" + + async def _get_resource(session, resource_id: str, columns: list) -> dict: + if resource_id not in tables_index_rows: + raise web.HTTPNotFound() + row = tables_index_rows[resource_id] + if row.get("deleted_at") is not None: + deleted_at = row["deleted_at"] + dataset_id = row.get("dataset_id") + message = ( + f"Resource {resource_id} has been permanently deleted on {deleted_at} " + "by its producer." + ) + if dataset_id: + message += ( + f" You can find more information about this resource at " + f"https://www.data.gouv.fr/datasets/{dataset_id}" + ) + else: + message += " Contact the resource producer to get more information." + raise web.HTTPGone(text=message) + + requested = set(columns) + record: dict[str, Any] = { + "deleted_at": row["deleted_at"], + "dataset_id": row["dataset_id"], + } + if "created_at" in requested: + record["created_at"] = row["created_at"] + if "url" in requested: + record["url"] = row["url"] + if "parsing_table" in requested: + record["parsing_table"] = row["parsing_table"] + if any(c.startswith("profile") or "csv_detective" in c for c in requested): + record["profile"] = json.loads(row["csv_detective"]) + return record + + mocker.patch("api_tabular.tabular.utils.get_resource", side_effect=_get_resource) + mocker.patch("api_tabular.tabular.app.get_resource", side_effect=_get_resource) + + +@pytest.fixture +def rmock(): + # passthrough for local requests (aiohttp TestServer) + with aioresponses(passthrough=["http://127.0.0.1"]) as m: + yield m + + +@pytest_asyncio.fixture +async def client() -> AsyncGenerator[TestClient, Any]: + app = await app_factory() + async with TestClient(TestServer(app)) as client: + yield client + + +@pytest.fixture +def base_url() -> Generator[str, Any, Any]: + yield f"{config.SCHEME}://{config.SERVER_NAME}" diff --git a/tests/test_api.py b/tests/test_api.py index 703db8a..178d93d 100644 --- a/tests/test_api.py +++ b/tests/test_api.py @@ -77,20 +77,12 @@ async def test_api_resource_meta_not_found(client): RESOURCE_ID, ], ) -async def test_api_resource_profile(client, tables_index_rows, exceptions_rows, _resource_id): - indexes = ( - list(json.loads(exceptions_rows[_resource_id]["table_indexes"]).keys()) - if _resource_id in exceptions_rows - else None - ) +async def test_api_resource_profile(client, tables_index_rows, _resource_id): res = await client.get(f"/api/resources/{_resource_id}/profile/") assert res.status == 200 body = await res.json() assert body["profile"] == json.loads(tables_index_rows[_resource_id]["csv_detective"]) - if indexes is None: - assert body["indexes"] is None - else: - assert sorted(body["indexes"]) == sorted(indexes) + assert "indexes" not in body async def test_api_resource_profile_not_found(client): @@ -285,75 +277,19 @@ async def test_api_with_unsupported_args(client): @pytest.mark.parametrize( - "params", - [ - (INDEXED_RESOURCE_ID, True), - (AGG_ALLOWED_INDEXED_RESOURCE_ID, False), - ], -) -async def test_api_exception_resource_indexes(client, tables_index_rows, exceptions_rows, params): - _resource_id, forbidden = params - detection = json.loads(tables_index_rows[_resource_id]["csv_detective"]) - indexes = list(json.loads(exceptions_rows[_resource_id]["table_indexes"]).keys()) - res = await client.get(f"/api/resources/{_resource_id}/profile/") - assert res.status == 200 - content = await res.json() - assert content["profile"] == detection - # sorted because it's made from a set so the order might not be preserved - assert sorted(indexes) == list(sorted(content["indexes"])) - - # checking that the resource is readable with no filter - res = await client.get(f"/api/resources/{_resource_id}/data/?page=1&page_size=1") - assert res.status == 200 - - # checking that the resource can be filtered on any columns - for col in detection["columns"].keys(): - if detection["columns"][col]["python_type"] == "json": - # can't handle json type for now - continue - res = await client.get( - f"/api/resources/{_resource_id}/data/?{col}__exact=1&page=1&page_size=1" - ) - assert res.status == 200 - - # checking that the resource cannot be aggregated on a non-indexed column - non_indexed_cols = [col for col in detection["columns"].keys() if col not in indexes] - for col in non_indexed_cols: - res = await client.get( - f"/api/resources/{_resource_id}/data/?{col}__groupby&page=1&page_size=1" - ) - assert res.status == 403 - - # checking whether aggregation is allowed on indexed columns - for idx in indexes: - res = await client.get( - f"/api/resources/{_resource_id}/data/?{idx}__groupby&page=1&page_size=1" - ) - assert res.status == 403 if forbidden else 200 - - -@pytest.mark.parametrize( - "params", + "_resource_id", [ - (RESOURCE_ID, True), - (AGG_ALLOWED_RESOURCE_ID, False), + INDEXED_RESOURCE_ID, + AGG_ALLOWED_INDEXED_RESOURCE_ID, + RESOURCE_ID, + AGG_ALLOWED_RESOURCE_ID, ], ) -async def test_api_exception_resource_no_indexes(client, tables_index_rows, params): - _resource_id, forbidden = params +async def test_api_resource_aggregations_allowed(client, tables_index_rows, _resource_id): detection = json.loads(tables_index_rows[_resource_id]["csv_detective"]) - # checking that we have an `indexes` key in the profile endpoint - res = await client.get(f"/api/resources/{_resource_id}/profile/") - assert res.status == 200 - content = await res.json() - assert content["profile"] == detection - assert content["indexes"] is None - - # checking that the resource is readable with no filter res = await client.get(f"/api/resources/{_resource_id}/data/?page=1&page_size=1") assert res.status == 200 - # checking that the resource can be filtered on all columns for col, results in detection["columns"].items(): if results["python_type"] == "json": continue @@ -361,16 +297,10 @@ async def test_api_exception_resource_no_indexes(client, tables_index_rows, para f"/api/resources/{_resource_id}/data/?{col}__exact=1&page=1&page_size=1" ) assert res.status == 200 - - # if aggregation is allowed: - # checking whether aggregation is allowed on all columns or none - for col, results in detection["columns"].items(): - if results["python_type"] == "json": - continue res = await client.get( f"/api/resources/{_resource_id}/data/?{col}__groupby&page=1&page_size=1" ) - assert res.status == 403 if forbidden else 200 + assert res.status == 200 @pytest.mark.parametrize( @@ -380,11 +310,23 @@ async def test_api_exception_resource_no_indexes(client, tables_index_rows, para (200, 200, ["status", "version", "uptime_since"]), ], ) -async def test_health(client, rmock, params): +async def test_health(client, mocker, params): postgrest_resp_code, api_expected_resp_code, expected_keys = params - rmock.head( - f"{config.PGREST_ENDPOINT}/migrations_csv", - status=postgrest_resp_code, + + class FakeResponse: + def __init__(self, ok: bool): + self.ok = ok + + async def __aenter__(self): + return self + + async def __aexit__(self, *args): + return None + + mocker.patch.object( + client.app["csession"], + "head", + return_value=FakeResponse(ok=postgrest_resp_code == 200), ) res = await client.get("/health/") assert res.status == api_expected_resp_code @@ -392,13 +334,6 @@ async def test_health(client, rmock, params): assert all(key in res_json for key in expected_keys) -async def test_aggregation_exceptions(client): - res = await client.get("/api/aggregation-exceptions/") - aggregations = await res.json() - assert not aggregations["allowed"] - assert aggregations["exceptions"] == config.ALLOW_AGGREGATION_EXCEPTIONS - - @pytest.mark.parametrize( "_resource_id,batch_size", [ diff --git a/tests/test_config.py b/tests/test_config.py index 8f00885..d2bd2ab 100644 --- a/tests/test_config.py +++ b/tests/test_config.py @@ -40,11 +40,12 @@ def test_custom_config_file_override(): def test_env_override(): os.environ["PGREST_ENDPOINT"] = "https://example.com" os.environ["PAGE_SIZE_MAX"] = "200" - os.environ["ALLOW_AGGREGATION"] = "true" - os.environ["ALLOW_AGGREGATION_EXCEPTIONS"] = "a,b,c" + os.environ["S3_ENDPOINT"] = "s3.example.com" + os.environ["S3_BUCKET"] = "my-bucket" config = Configurator() assert config.PGREST_ENDPOINT == "https://example.com" assert config.PAGE_SIZE_MAX == 200 - assert config.ALLOW_AGGREGATION - assert config.ALLOW_AGGREGATION_EXCEPTIONS == ["a", "b", "c"] + assert config.S3_ENDPOINT == "s3.example.com" + assert config.S3_BUCKET == "my-bucket" + assert config.parquet_url("abc") == "https://s3.example.com/my-bucket/parquet/abc.parquet" diff --git a/tests/test_duckdb_query.py b/tests/test_duckdb_query.py new file mode 100644 index 0000000..8f0cf69 --- /dev/null +++ b/tests/test_duckdb_query.py @@ -0,0 +1,51 @@ +from api_tabular.tabular.duckdb_query import build_duckdb_query + + +def test_duckdb_query_limit_and_default_order(): + q = build_duckdb_query([], page_size=12) + assert q.limit == 12 + assert q.offset == 0 + assert q.order_sql == 'ORDER BY "__id" ASC' + assert q.select_sql == "*" + assert not q.is_aggregation + + +def test_duckdb_query_filters(): + q = build_duckdb_query( + ["score__greater=0.9", "decompte__exact=13"], + page_size=20, + offset=20, + ) + assert 'WHERE "score" >= ? AND "decompte" = ?' == q.where_sql + assert q.where_params == ["0.9", "13"] + assert q.limit == 20 + assert q.offset == 20 + + +def test_duckdb_query_contains_and_or(): + q = build_duckdb_query( + ["or=(id__exact.abc,score__less.0.1)"], + page_size=50, + ) + assert q.where_sql.startswith("WHERE (") + assert " OR " in q.where_sql + assert q.where_params == ["abc", "0.1"] + + +def test_duckdb_query_aggregation(): + q = build_duckdb_query( + ["decompte__groupby", "score__avg"], + page_size=50, + ) + assert q.is_aggregation + assert '"decompte"' in q.select_sql + assert 'AVG("score") AS "score__avg"' in q.select_sql + assert q.group_by_sql == 'GROUP BY "decompte"' + assert q.order_sql == "" + + +def test_duckdb_query_columns_conflict_with_agg(): + import pytest + + with pytest.raises(ValueError): + build_duckdb_query(["columns=a,b", "a__groupby"]) diff --git a/tests/test_query.py b/tests/test_query.py index 69f75ff..ed0bac2 100644 --- a/tests/test_query.py +++ b/tests/test_query.py @@ -214,17 +214,11 @@ def test_query_build_multiple_with_unknown(): ], ) def test_query_aggregators(allow_aggregation, mocker): - if allow_aggregation: - mocker.patch("api_tabular.config.ALLOW_AGGREGATION_EXCEPTIONS", [RESOURCE_ID]) query_str = [ "column_name__groupby", "column_name__min", "column_name__avg", ] - if not allow_aggregation: - with pytest.raises(PermissionError): - build_sql_query_string(query_str, resource_id=RESOURCE_ID, page_size=50) - return results = build_sql_query_string(query_str, resource_id=RESOURCE_ID, page_size=50).split("&") assert "limit=50" in results assert "order=__id.asc" not in results # no sort if aggregators diff --git a/tests/test_swagger.py b/tests/test_swagger.py index 68f8e08..99eb213 100644 --- a/tests/test_swagger.py +++ b/tests/test_swagger.py @@ -30,14 +30,17 @@ async def test_swagger_endpoint(client, _resource_id): @pytest.mark.parametrize( - "params", + "_resource_id", [ - (RESOURCE_ID, False), - (AGG_ALLOWED_RESOURCE_ID, True), + RESOURCE_ID, + AGG_ALLOWED_RESOURCE_ID, + INDEXED_RESOURCE_ID, + AGG_ALLOWED_INDEXED_RESOURCE_ID, ], ) -async def test_swagger_no_indexes(client, tables_index_rows, params): - _resource_id, allow_aggregation = params +async def test_swagger_includes_all_columns_and_aggregators( + client, tables_index_rows, _resource_id +): detection = json.loads(tables_index_rows[_resource_id]["csv_detective"]) columns = {c: v["python_type"] for c, v in detection["columns"].items()} res = await client.get(f"/api/resources/{_resource_id}/swagger/") @@ -58,46 +61,9 @@ async def test_swagger_no_indexes(client, tables_index_rows, params): else [p] ) for _p in _params: - if allow_aggregation: - if f"{c}__{_p}" not in params: - missing.append(f"{c}__{_p} is missing in {output} output") - elif OPERATORS_DESCRIPTIONS.get(_p, {}).get("is_aggregator"): - assert params[f"{c}__{_p}"].get("allowEmptyValue") - else: - if ( - not OPERATORS_DESCRIPTIONS.get(_p, {}).get("is_aggregator") - and f"{c}__{_p}" not in params # filters are in - ): - missing.append(f"{c}__{_p} is missing in {output} output") - assert params[f"{c}__{_p}"].get("allowEmptyValue") is None - if ( - OPERATORS_DESCRIPTIONS.get(_p, {}).get("is_aggregator") - and f"{c}__{_p}" in params # aggregators are out - ): - missing.append(f"{c}__{_p} is in {output} output but should not") + if f"{c}__{_p}" not in params: + missing.append(f"{c}__{_p} is missing in {output} output") + elif OPERATORS_DESCRIPTIONS.get(_p, {}).get("is_aggregator"): + assert params[f"{c}__{_p}"].get("allowEmptyValue") if missing: raise ValueError("\n" + ";\n".join(missing)) - - -@pytest.mark.parametrize( - "_resource_id", - [ - AGG_ALLOWED_INDEXED_RESOURCE_ID, - INDEXED_RESOURCE_ID, - ], -) -async def test_swagger_with_indexes(client, tables_index_rows, exceptions_rows, _resource_id): - detection = json.loads(tables_index_rows[_resource_id]["csv_detective"]) - indexes = list(json.loads(exceptions_rows[_resource_id]["table_indexes"]).keys()) - non_indexed_cols = [col for col in detection["columns"].keys() if col not in indexes] - res = await client.get(f"/api/resources/{_resource_id}/swagger/") - swagger = await res.text() - swagger_dict = yaml.safe_load(swagger) - - for output in ["json", "csv"]: - params = swagger_dict["paths"][ - f"/api/resources/{_resource_id}/data/{'' if output == 'json' else 'csv/'}" - ]["parameters"] - params = set([p["name"].split("__")[0] for p in params if "__" in p["name"]]) - assert all(c in params for c in indexes) - assert not any(c in params for c in non_indexed_cols) diff --git a/uv.lock b/uv.lock index 35895ad..78f9e6f 100644 --- a/uv.lock +++ b/uv.lock @@ -263,6 +263,42 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/d1/ae/afb1487556e2dc827a17097aac8158a25b433a345386f0e249f6d2694ccb/devtools-0.12.2-py3-none-any.whl", hash = "sha256:c366e3de1df4cdd635f1ad8cbcd3af01a384d7abda71900e68d43b04eb6aaca7", size = 19411, upload-time = "2023-09-03T16:56:59.049Z" }, ] +[[package]] +name = "duckdb" +version = "1.5.4" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/31/29/9bad86ed7aa812d8c822a27c15c355b6d5423b991feeec86ed18027b6daa/duckdb-1.5.4.tar.gz", hash = "sha256:f9e32f1cdd106793d79d190186bed9e75289d51e68bd9174e47c04bffedeab6f", size = 18046634, upload-time = "2026-06-17T10:48:52.499Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/56/bb/7921dabd50daef3969f14cd8a5a14c24eee337db7914a462f2defa8add92/duckdb-1.5.4-cp311-cp311-macosx_10_9_universal2.whl", hash = "sha256:3fb41d9cfccb7e44511eeeed263ae98143ca63bdb1ef84631ba637c314efa1b5", size = 32663142, upload-time = "2026-06-17T10:47:45.471Z" }, + { url = "https://files.pythonhosted.org/packages/a6/83/2137765eaba6a9aefe3bb9848ddaac7407fe3ba19b292f98b31f3b7ab27f/duckdb-1.5.4-cp311-cp311-macosx_10_9_x86_64.whl", hash = "sha256:8ba7b666bc9c78d6a930ee9f469024149f0c6a23fb7d2c3418aad6774339bec0", size = 17321485, upload-time = "2026-06-17T10:47:47.778Z" }, + { url = "https://files.pythonhosted.org/packages/0e/b2/a02c1ee43fd7e8cf1fc2e3d377f3dcf9d4a3e58a4549557516e1866ff0da/duckdb-1.5.4-cp311-cp311-macosx_11_0_arm64.whl", hash = "sha256:9d9e6817fcbc09d2605a2c8c041ac7824d738d917c35a4d427e977647e1d7944", size = 15470820, upload-time = "2026-06-17T10:47:49.977Z" }, + { url = "https://files.pythonhosted.org/packages/d8/48/a243d30223b024bc6057abe472b002cff01e97efefb4d2f0b0dcc5aece0b/duckdb-1.5.4-cp311-cp311-manylinux_2_26_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:02dd9f9a6124069213f13e3a474c208028c472fe1acdae12b38761f954fe4fc6", size = 19341849, upload-time = "2026-06-17T10:47:52.205Z" }, + { url = "https://files.pythonhosted.org/packages/08/ff/a5d48de4771e2403a8ef26a20dc7457b1c8f7e398ff0caf9c0cad8805f89/duckdb-1.5.4-cp311-cp311-manylinux_2_26_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:ccc7f2694d02b4763fee61021d45e12f7bc5743993686563957df0cef799fbae", size = 21451698, upload-time = "2026-06-17T10:47:54.653Z" }, + { url = "https://files.pythonhosted.org/packages/79/b8/8244d7741b4afae67775cf0cb0d4eb9e923a83110907e4801e17fa078480/duckdb-1.5.4-cp311-cp311-win_amd64.whl", hash = "sha256:4c430e788d99b50854209bf2833ba36a45df75e57f86efb477046cd408bbd077", size = 13132643, upload-time = "2026-06-17T10:47:56.75Z" }, + { url = "https://files.pythonhosted.org/packages/e4/57/8169822a37f6dd7d561c567f9007e3cf04bf97bccb619afe90db849c0962/duckdb-1.5.4-cp311-cp311-win_arm64.whl", hash = "sha256:e2dc8340cfb6006025a798c50f40126d6e945a1d2487be94667bb4166556ce7b", size = 13986386, upload-time = "2026-06-17T10:47:59.345Z" }, + { url = "https://files.pythonhosted.org/packages/c8/f2/e2f4b477ae3a3b40e8b5f429832e48edb62ed9da99807cc4902e157e5646/duckdb-1.5.4-cp312-cp312-macosx_10_13_universal2.whl", hash = "sha256:291a9e7502551170af989ff63139a7a49e99d68edbc5ef5017ac27541fe54c65", size = 32708876, upload-time = "2026-06-17T10:48:01.527Z" }, + { url = "https://files.pythonhosted.org/packages/2e/2b/b698d82a5e1e30b6a05748d72045f672994c6b22f4f0f8423523608b991f/duckdb-1.5.4-cp312-cp312-macosx_10_13_x86_64.whl", hash = "sha256:83e8c089bbb756ca4471d8b05943b80a106058697cf00615e70423106bb783bc", size = 17346125, upload-time = "2026-06-17T10:48:04.035Z" }, + { url = "https://files.pythonhosted.org/packages/71/75/37e13f39268eaf34864453b3a039c4a1ff0b088d3eae45a4289b41c98c1b/duckdb-1.5.4-cp312-cp312-macosx_11_0_arm64.whl", hash = "sha256:ff96d2a342b200e1ec6f1f19986c77f4ac16a49b6112f71c5b763989203a9d60", size = 15488133, upload-time = "2026-06-17T10:48:06.312Z" }, + { url = "https://files.pythonhosted.org/packages/cc/59/2d082af578f689231798245b54562c61416e49049b0bda81a06c56a4b53e/duckdb-1.5.4-cp312-cp312-manylinux_2_26_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:8f935ef210ab00bc94bb1e3052697adaa36bb0ce7bdfeda8b0f34e2ff1643870", size = 19367895, upload-time = "2026-06-17T10:48:08.59Z" }, + { url = "https://files.pythonhosted.org/packages/52/2b/55c34d2863a76ca824ef8274691e84240b4ff1acde3d231709e82557c240/duckdb-1.5.4-cp312-cp312-manylinux_2_26_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:0cda263d8c20addb8d4f95464787cbe0af1144f7ab7e21db3709fb826ee01725", size = 21486499, upload-time = "2026-06-17T10:48:10.963Z" }, + { url = "https://files.pythonhosted.org/packages/cf/30/ade5952b8182fac86fab43b95ebe3836e66381d0ad64eb1e54bd8207c988/duckdb-1.5.4-cp312-cp312-win_amd64.whl", hash = "sha256:266c7c909558ce7377f57d082cee408aadebdd9111be017558ca54e44a031037", size = 13147934, upload-time = "2026-06-17T10:48:13.061Z" }, + { url = "https://files.pythonhosted.org/packages/f5/00/278f0f70e25b9911afe2fd227b9460f2e6d76177f0dcc03f7f1454afefa5/duckdb-1.5.4-cp312-cp312-win_arm64.whl", hash = "sha256:f14e79a006341f29ee5a2692a24dac5114e77533d579c57ec39124adf0135033", size = 13965235, upload-time = "2026-06-17T10:48:15.782Z" }, + { url = "https://files.pythonhosted.org/packages/da/69/3fcb34e523a9bad1f0557a6c7691a71ba66c43a05e5be9ee96a9a841ed65/duckdb-1.5.4-cp313-cp313-macosx_10_13_universal2.whl", hash = "sha256:42a612e67d64450b446eb69695290d460713eef46e0f64467ab9dfe96264ee05", size = 32708366, upload-time = "2026-06-17T10:48:18.084Z" }, + { url = "https://files.pythonhosted.org/packages/f5/5f/bff5054c2c1d65decab36aa6296621e51a2a575a9f250db7ab9b83a325d6/duckdb-1.5.4-cp313-cp313-macosx_10_13_x86_64.whl", hash = "sha256:3fb6f07d54ecf4d0d3c5179a2361fdddfafa14de4fc42696de4632479b703421", size = 17345735, upload-time = "2026-06-17T10:48:20.67Z" }, + { url = "https://files.pythonhosted.org/packages/93/12/d1b2b344e9699246aada6f9de5156e708fb476e2780e5bff9b5d95fe11d9/duckdb-1.5.4-cp313-cp313-macosx_11_0_arm64.whl", hash = "sha256:0f32ad7e0286c1c29ab6c73b29118c86101f8eee46aae54f54d0b50916f542f6", size = 15488568, upload-time = "2026-06-17T10:48:23.038Z" }, + { url = "https://files.pythonhosted.org/packages/c1/d1/ac56c6096e3e95da60b2c5dd5a0f0eb5540a80622e2e4f8faab893ec4e96/duckdb-1.5.4-cp313-cp313-manylinux_2_26_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:698ec90bd5d5538bd5f6d212a4b61af443d240703cf45f134738535026556ea5", size = 19368184, upload-time = "2026-06-17T10:48:25.601Z" }, + { url = "https://files.pythonhosted.org/packages/1e/0b/2ae4c3e157a19d9b4ac1f09a5dea6f93012334cc2db09f1e0c71eb99693d/duckdb-1.5.4-cp313-cp313-manylinux_2_26_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:136cea7f886b78caf4035485b4b1e766e8b309e999f9e83a966f81ebb8122844", size = 21486523, upload-time = "2026-06-17T10:48:27.817Z" }, + { url = "https://files.pythonhosted.org/packages/64/7b/c3d8d21e0d0db8faa81eeeb3a55b9932f5a0a16466cb968dc713a653d701/duckdb-1.5.4-cp313-cp313-win_amd64.whl", hash = "sha256:bd6777e8ddd74fb603a6d09766bfcff28638189f8aaa61fc0dffd9e9a4baa8e5", size = 13147807, upload-time = "2026-06-17T10:48:30.017Z" }, + { url = "https://files.pythonhosted.org/packages/44/48/ddf8d3740e3d28582944f70d84e720b5dc28c10ec22b668a0e0bd965f2f2/duckdb-1.5.4-cp313-cp313-win_arm64.whl", hash = "sha256:73f4878a3012283024a64a1909e440aac12091ef336f671fc142f7e87449ce0c", size = 13965189, upload-time = "2026-06-17T10:48:32.251Z" }, + { url = "https://files.pythonhosted.org/packages/62/01/67ac4cbc8e552a1e14c029b5c443d828e68f94d5d913c574f577e1db277e/duckdb-1.5.4-cp314-cp314-macosx_10_15_universal2.whl", hash = "sha256:4647968629d0677bbcc2416c7aeda8685eb84e4ca15a6dbd4f82a66cfc91a532", size = 32714364, upload-time = "2026-06-17T10:48:34.724Z" }, + { url = "https://files.pythonhosted.org/packages/4e/0e/eb44d983fa56b175f971eea251bde284a36d26cbb93fcb68035061f54078/duckdb-1.5.4-cp314-cp314-macosx_10_15_x86_64.whl", hash = "sha256:e8fcef301cf68d3951ea1eb8ac4d76cea0a6f6a08f4c78fe4026fc96d217bebc", size = 17349820, upload-time = "2026-06-17T10:48:37.126Z" }, + { url = "https://files.pythonhosted.org/packages/10/b2/b9dc7624b105d414585b8530451c1162c0b4750c0be9be2e497bb47a8a9b/duckdb-1.5.4-cp314-cp314-macosx_11_0_arm64.whl", hash = "sha256:f6f39cd0dc6948dee17fd130aec55114f97a8ef6e1db519b9774087962bc5c8c", size = 15498160, upload-time = "2026-06-17T10:48:40.032Z" }, + { url = "https://files.pythonhosted.org/packages/b7/57/61356444f6a8c62dec3c3d129abfc53f428de1d484093d1bb381db441231/duckdb-1.5.4-cp314-cp314-manylinux_2_26_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:262f068158beb5943f2c618f4e54b46db8306b959f90dce956f90a89f613673d", size = 19374183, upload-time = "2026-06-17T10:48:42.698Z" }, + { url = "https://files.pythonhosted.org/packages/b0/f4/d5d633dd7c5138d8f7c434e6ac2553c831b7fb658494efa8d0bc73df8623/duckdb-1.5.4-cp314-cp314-manylinux_2_26_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:4d2307a76d199077b0055b354e90e857479461a0d875437535dd4833172c8b6d", size = 21487202, upload-time = "2026-06-17T10:48:45.407Z" }, + { url = "https://files.pythonhosted.org/packages/c0/26/5be13bbd5c3421dccfc1ad4ca9da4b97c5a3ddd73f66542092f3167ec52c/duckdb-1.5.4-cp314-cp314-win_amd64.whl", hash = "sha256:6dcbb81a1276bc48deb4d562bce4f8895e4fc6348750a096e30052345c6d6552", size = 13666989, upload-time = "2026-06-17T10:48:47.764Z" }, + { url = "https://files.pythonhosted.org/packages/dc/82/4d52f3f9f9703a226b26b80bdae3f6905aeefe5221bf1815fc93ff02ca25/duckdb-1.5.4-cp314-cp314-win_arm64.whl", hash = "sha256:0f8722346024e5d9f02b58bf7b0491a629f97fdc8a04a10e432940f471ee387a", size = 14449863, upload-time = "2026-06-17T10:48:50.18Z" }, +] + [[package]] name = "executing" version = "2.2.1" @@ -917,6 +953,7 @@ source = { editable = "." } dependencies = [ { name = "aiohttp" }, { name = "aiohttp-swagger" }, + { name = "duckdb" }, { name = "gunicorn" }, { name = "sentry-sdk" }, ] @@ -936,6 +973,7 @@ dev = [ requires-dist = [ { name = "aiohttp", specifier = ">=3.14.1,<4.0.0" }, { name = "aiohttp-swagger", specifier = ">=1.0.16,<2.0.0" }, + { name = "duckdb", specifier = ">=1.5.4" }, { name = "gunicorn", specifier = ">=23.0.0,<24.0.0" }, { name = "sentry-sdk", specifier = ">=2.49.0,<3.0.0" }, ]