diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml new file mode 100644 index 0000000..8357ac3 --- /dev/null +++ b/.github/workflows/ci.yml @@ -0,0 +1,64 @@ +name: CI + +on: + push: + branches: [main] + pull_request: + +jobs: + lint: + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v4 + + - uses: actions/setup-python@v5 + with: + python-version: "3.13" + + - name: Install uv + uses: astral-sh/setup-uv@v5 + + - name: Install dependencies + run: uv sync --group dev + + - name: Run pre-commit + uses: pre-commit/action@v3.0.1 + + test: + runs-on: ubuntu-latest + strategy: + matrix: + python-version: ["3.11", "3.12", "3.13"] + steps: + - uses: actions/checkout@v4 + + - uses: actions/setup-python@v5 + with: + python-version: ${{ matrix.python-version }} + + - name: Install uv + uses: astral-sh/setup-uv@v5 + + - name: Install dependencies + run: uv sync --group dev + + - name: Run unit tests + run: uv run pytest tests/test_client.py tests/test_dbapi.py -v + + integration: + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v4 + + - uses: actions/setup-python@v5 + with: + python-version: "3.13" + + - name: Install uv + uses: astral-sh/setup-uv@v5 + + - name: Install dependencies + run: uv sync --group dev + + - name: Run integration tests + run: uv run pytest tests/test_integration.py -v -s -m integration diff --git a/.github/workflows/publish.yml b/.github/workflows/publish.yml new file mode 100644 index 0000000..05804cf --- /dev/null +++ b/.github/workflows/publish.yml @@ -0,0 +1,29 @@ +name: Publish to PyPI + +on: + push: + tags: + - "v*" + +jobs: + publish: + runs-on: ubuntu-latest + permissions: + id-token: write # Required for trusted publishing + steps: + - uses: actions/checkout@v4 + with: + fetch-depth: 0 # Full history needed for hatch-vcs + + - uses: actions/setup-python@v5 + with: + python-version: "3.13" + + - name: Install build tools + run: pip install build + + - name: Build package + run: python -m build + + - name: Publish to PyPI + uses: pypa/gh-action-pypi-publish@release/v1 diff --git a/.gitignore b/.gitignore index 6b562ea..fbb8d77 100644 --- a/.gitignore +++ b/.gitignore @@ -9,3 +9,6 @@ wheels/ # Virtual environments .venv .idea + +# Auto-generated version file +src/flink_gateway/_version.py diff --git a/README.md b/README.md index 432e3ee..bb91603 100644 --- a/README.md +++ b/README.md @@ -1,5 +1,9 @@ # py-flink-sql-gateway +[![CI](https://github.com/exness/py-flink-sql-gateway/actions/workflows/ci.yml/badge.svg)](https://github.com/exness/py-flink-sql-gateway/actions/workflows/ci.yml) +[![PyPI version](https://img.shields.io/pypi/v/py-flink-sql-gateway)](https://pypi.org/project/py-flink-sql-gateway/) +[![Python](https://img.shields.io/pypi/pyversions/py-flink-sql-gateway)](https://pypi.org/project/py-flink-sql-gateway/) + A lightweight Python driver for the **Apache Flink SQL Gateway**, implementing [PEP 249 (DB-API 2.0)](https://peps.python.org/pep-0249/). ## Installation @@ -61,6 +65,18 @@ with connect("http://localhost:8083") as conn: > **Tip:** `ROW` and `MAP` types arrive as Python `dict`, `ARRAY` arrives as `list`. Binary data is passed through as-is. +### Connection options + +```python +# Pass Flink session properties +with connect("http://localhost:8083", properties={"pipeline.name": "my-job"}) as conn: + ... + +# Set a query timeout (default: 300s) +with conn.cursor(query_timeout=60.0) as cur: + cur.execute("SELECT ...") +``` + ### Low-level REST access If you need features beyond DB-API, use the exported client directly: @@ -78,7 +94,7 @@ with FlinkSqlGatewayClient("http://localhost:8083") as client: Python uses `None` for SQL NULLs natively — no wrapper types needed. | Flink Type | Python Type | -|---------------------------|--------------------| +|---------------------------|:------------------:| | TINYINT / SMALLINT / INT | `int` | | BIGINT / INTERVAL | `int` | | FLOAT / DOUBLE | `float` | @@ -93,21 +109,29 @@ Python uses `None` for SQL NULLs natively — no wrapper types needed. | MAP | `dict` | | ARRAY | `list` | +Nested complex types (e.g. `MAP>>>`) are recursively decoded to the correct Python types. + --- ## Development & tests -Requires Python 3.13+ and [uv](https://docs.astral.sh/uv/). +Requires Python 3.11+ and [uv](https://docs.astral.sh/uv/). ```bash # Install dependencies -uv sync +uv sync --group dev + +# Set up pre-commit hooks (runs on every push) +pre-commit install --hook-type pre-push # Run unit tests (no Docker needed) uv run pytest tests/test_client.py tests/test_dbapi.py -v -# Run all tests including integration (requires Docker) -uv run pytest tests/ -v +# Run integration tests (requires Docker) +uv run pytest tests/test_integration.py -v -s -m integration + +# Run all pre-commit checks manually +pre-commit run --all-files ``` Integration tests spin up a Flink cluster (JobManager + TaskManager + SQL Gateway) via [testcontainers-python](https://testcontainers-python.readthedocs.io/). @@ -116,4 +140,4 @@ Integration tests spin up a Flink cluster (JobManager + TaskManager + SQL Gatewa ## License -MIT (see LICENSE) +MIT diff --git a/pyproject.toml b/pyproject.toml index 1c43544..224b1d3 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,12 +1,18 @@ [project] name = "py-flink-sql-gateway" -version = "0.1.0" -description = "A Python client for the Apache Flink SQL Gateway REST API" +dynamic = ["version"] +description = "A lightweight Python driver for the Apache Flink SQL Gateway, implementing PEP 249 (DB-API 2.0)." readme = "README.md" authors = [ { name = "Ilya Soin", email = "ilya.soin@exness.com" } ] requires-python = ">=3.11" +classifiers = [ + "Programming Language :: Python :: 3", + "Programming Language :: Python :: 3.11", + "Programming Language :: Python :: 3.12", + "Programming Language :: Python :: 3.13", +] dependencies = [ "httpx>=0.28.1", ] @@ -18,9 +24,16 @@ dev = [ ] [build-system] -requires = ["hatchling"] +requires = ["hatchling", "hatch-vcs"] build-backend = "hatchling.build" +[tool.hatch.version] +source = "vcs" +raw-options = { version_scheme = "guess-next-dev" } + +[tool.hatch.build.hooks.vcs] +version-file = "src/flink_gateway/_version.py" + [tool.hatch.build.targets.wheel] packages = ["src/flink_gateway"] diff --git a/src/flink_gateway/models.py b/src/flink_gateway/models.py index 99d63dc..b6190ad 100644 --- a/src/flink_gateway/models.py +++ b/src/flink_gateway/models.py @@ -112,12 +112,36 @@ class LogicalType: precision: int | None = None scale: int | None = None resolution: str | None = None - children: list[ColumnInfo] = field(default_factory=list) + # ARRAY element type + element_type: LogicalType | None = None + # MAP key/value types + key_type: LogicalType | None = None + value_type: LogicalType | None = None + # ROW field descriptors + fields: list[ColumnInfo] = field(default_factory=list) @classmethod def from_dict(cls, data: dict[str, Any]) -> LogicalType: - children_raw = data.get("children", []) - children = [ColumnInfo.from_dict(c) for c in children_raw] + element_type = None + key_type = None + value_type = None + fields: list[ColumnInfo] = [] + + if "elementType" in data: + element_type = LogicalType.from_dict(data["elementType"]) + if "keyType" in data: + key_type = LogicalType.from_dict(data["keyType"]) + if "valueType" in data: + value_type = LogicalType.from_dict(data["valueType"]) + if "fields" in data: + for f in data["fields"]: + fields.append( + ColumnInfo( + name=f.get("name", ""), + logical_type=LogicalType.from_dict(f.get("fieldType", {})), + ) + ) + return cls( type=data.get("type", ""), nullable=data.get("nullable", True), @@ -125,7 +149,10 @@ def from_dict(cls, data: dict[str, Any]) -> LogicalType: precision=data.get("precision"), scale=data.get("scale"), resolution=data.get("resolution"), - children=children, + element_type=element_type, + key_type=key_type, + value_type=value_type, + fields=fields, ) diff --git a/src/flink_gateway/types.py b/src/flink_gateway/types.py index 4031163..3251774 100644 --- a/src/flink_gateway/types.py +++ b/src/flink_gateway/types.py @@ -224,7 +224,7 @@ def decode_field( value: The raw JSON value from the gateway response. flink_type: The normalized Flink type. precision: Optional precision from column metadata. - logical_type: Optional full ``LogicalType`` with children for + logical_type: Optional full ``LogicalType`` with element_type/fields for recursive decoding of ROW / MAP types. Returns: @@ -314,18 +314,22 @@ def _decode_array_value( value: Any, logical_type: LogicalType | None, ) -> Any: - """Decode an ARRAY value using its element child schema.""" - if not isinstance(value, list) or logical_type is None or not logical_type.children: + """Decode an ARRAY value using its element type schema.""" + if ( + not isinstance(value, list) + or logical_type is None + or logical_type.element_type is None + ): return value - elem_child = logical_type.children[0] - elem_ft = normalize_flink_type(elem_child.logical_type.type) + elem_lt = logical_type.element_type + elem_ft = normalize_flink_type(elem_lt.type) return [ decode_field( item, elem_ft, - precision=elem_child.logical_type.precision, - logical_type=elem_child.logical_type, + precision=elem_lt.precision, + logical_type=elem_lt, ) for item in value ] @@ -335,21 +339,21 @@ def _decode_row_value( value: Any, logical_type: LogicalType | None, ) -> Any: - """Decode a ROW value using its child schema.""" - if not isinstance(value, dict) or logical_type is None or not logical_type.children: + """Decode a ROW value using its field schema.""" + if not isinstance(value, dict) or logical_type is None or not logical_type.fields: return value - children_by_name = {child.name: child for child in logical_type.children} + fields_by_name = {f.name: f for f in logical_type.fields} decoded: dict[str, Any] = {} for key, raw_val in value.items(): - child = children_by_name.get(key) - if child is not None: - child_ft = normalize_flink_type(child.logical_type.type) + fld = fields_by_name.get(key) + if fld is not None: + fld_ft = normalize_flink_type(fld.logical_type.type) decoded[key] = decode_field( raw_val, - child_ft, - precision=child.logical_type.precision, - logical_type=child.logical_type, + fld_ft, + precision=fld.logical_type.precision, + logical_type=fld.logical_type, ) else: decoded[key] = raw_val @@ -360,32 +364,33 @@ def _decode_map_value( value: Any, logical_type: LogicalType | None, ) -> Any: - """Decode a MAP value using its key/value child schemas.""" + """Decode a MAP value using its key/value type schemas.""" if ( not isinstance(value, dict) or logical_type is None - or len(logical_type.children) < 2 + or logical_type.key_type is None + or logical_type.value_type is None ): return value - key_child = logical_type.children[0] - val_child = logical_type.children[1] - key_ft = normalize_flink_type(key_child.logical_type.type) - val_ft = normalize_flink_type(val_child.logical_type.type) + key_lt = logical_type.key_type + val_lt = logical_type.value_type + key_ft = normalize_flink_type(key_lt.type) + val_ft = normalize_flink_type(val_lt.type) decoded: dict[Any, Any] = {} for raw_key, raw_val in value.items(): dk = decode_field( raw_key, key_ft, - precision=key_child.logical_type.precision, - logical_type=key_child.logical_type, + precision=key_lt.precision, + logical_type=key_lt, ) dv = decode_field( raw_val, val_ft, - precision=val_child.logical_type.precision, - logical_type=val_child.logical_type, + precision=val_lt.precision, + logical_type=val_lt, ) decoded[dk] = dv return decoded diff --git a/tests/test_dbapi.py b/tests/test_dbapi.py index a4134d8..3b3bf4a 100644 --- a/tests/test_dbapi.py +++ b/tests/test_dbapi.py @@ -425,12 +425,12 @@ class TestRecursiveDecoding: """Tests for recursive ROW and MAP decoding.""" def test_row_decodes_children(self): - """A ROW with INT + STRING children should decode field values.""" + """A ROW with INT + STRING fields should decode field values.""" from flink_gateway.models import ColumnInfo, LogicalType lt = LogicalType( type="ROW", - children=[ + fields=[ ColumnInfo(name="score", logical_type=LogicalType(type="INTEGER")), ColumnInfo(name="label", logical_type=LogicalType(type="VARCHAR")), ], @@ -445,13 +445,13 @@ def test_row_nested(self): inner_lt = LogicalType( type="ROW", - children=[ + fields=[ ColumnInfo(name="x", logical_type=LogicalType(type="DOUBLE")), ], ) outer_lt = LogicalType( type="ROW", - children=[ + fields=[ ColumnInfo(name="id", logical_type=LogicalType(type="INTEGER")), ColumnInfo(name="nested", logical_type=inner_lt), ], @@ -466,14 +466,12 @@ def test_row_nested(self): def test_map_decodes_values(self): """A MAP should decode string values to int.""" - from flink_gateway.models import ColumnInfo, LogicalType + from flink_gateway.models import LogicalType lt = LogicalType( type="MAP", - children=[ - ColumnInfo(name="KEY", logical_type=LogicalType(type="VARCHAR")), - ColumnInfo(name="VALUE", logical_type=LogicalType(type="INTEGER")), - ], + key_type=LogicalType(type="VARCHAR"), + value_type=LogicalType(type="INTEGER"), ) ft = normalize_flink_type("MAP") result = decode_field({"apples": "3", "bananas": "5"}, ft, logical_type=lt) @@ -485,16 +483,14 @@ def test_map_with_row_values(self): row_lt = LogicalType( type="ROW", - children=[ + fields=[ ColumnInfo(name="score", logical_type=LogicalType(type="INTEGER")), ], ) lt = LogicalType( type="MAP", - children=[ - ColumnInfo(name="KEY", logical_type=LogicalType(type="VARCHAR")), - ColumnInfo(name="VALUE", logical_type=row_lt), - ], + key_type=LogicalType(type="VARCHAR"), + value_type=row_lt, ) ft = normalize_flink_type("MAP") result = decode_field( @@ -528,7 +524,7 @@ def test_row_null_value(self): lt = LogicalType( type="ROW", - children=[ + fields=[ ColumnInfo(name="x", logical_type=LogicalType(type="INTEGER")), ], ) @@ -537,13 +533,11 @@ def test_row_null_value(self): def test_array_decodes_elements(self): """An ARRAY should decode each element.""" - from flink_gateway.models import ColumnInfo, LogicalType + from flink_gateway.models import LogicalType lt = LogicalType( type="ARRAY", - children=[ - ColumnInfo(name="ELEMENT", logical_type=LogicalType(type="INTEGER")), - ], + element_type=LogicalType(type="INTEGER"), ) ft = normalize_flink_type("ARRAY") result = decode_field(["1", "2", "3"], ft, logical_type=lt) @@ -555,16 +549,14 @@ def test_array_of_rows(self): row_lt = LogicalType( type="ROW", - children=[ + fields=[ ColumnInfo(name="name", logical_type=LogicalType(type="VARCHAR")), ColumnInfo(name="score", logical_type=LogicalType(type="INTEGER")), ], ) lt = LogicalType( type="ARRAY", - children=[ - ColumnInfo(name="ELEMENT", logical_type=row_lt), - ], + element_type=row_lt, ) ft = normalize_flink_type("ARRAY") result = decode_field( diff --git a/tests/test_integration.py b/tests/test_integration.py index 5e83075..cd5a34f 100644 --- a/tests/test_integration.py +++ b/tests/test_integration.py @@ -98,8 +98,6 @@ def test_select_all_types_not_null(flink_gateway_url: str): cur.execute("SELECT * FROM test_types_nn") row = cur.fetchone() - print(row) - assert row is not None, "Expected at least one row from datagen" # Validate Python types @@ -333,11 +331,23 @@ def test_filesystem_row_complex_type(flink_gateway_url: str): id BIGINT NOT NULL, name STRING NOT NULL, info ROW NOT NULL, - nested_row ROW> NOT NULL, + nested_row ROW> NOT NULL, tags MAP NOT NULL, player_map MAP> NOT NULL, ts_array ARRAY NOT NULL, - row_array ARRAY> NOT NULL + row_array ARRAY> NOT NULL, + deep_map MAP> + >> NOT NULL, + price DECIMAL(10, 2) NOT NULL, + birth_date DATE NOT NULL, + start_time TIME(0) NOT NULL, + created_at TIMESTAMP(3) NOT NULL, + updated_ltz TIMESTAMP_LTZ(3) NOT NULL ) WITH ( 'connector' = 'filesystem', 'format' = 'json', @@ -360,7 +370,22 @@ def test_filesystem_row_complex_type(flink_gateway_url: str): TIMESTAMP '2024-01-01 10:00:00.000', TIMESTAMP '2024-06-15 14:30:00.000' ], - ARRAY[ROW('x', 10), ROW('y', 20)] + ARRAY[ROW('x', 10), ROW('y', 20)], + MAP[ + 'env1', + ROW( + 'prod', + MAP[ + 'svc1', + ROW(TIMESTAMP '2024-03-15 09:30:00.000') + ] + ) + ], + CAST(99.95 AS DECIMAL(10, 2)), + DATE '2024-06-15', + TIME '14:30:00', + TIMESTAMP '2024-06-15 14:30:00.123', + TO_TIMESTAMP_LTZ(1718458200123, 3) ), (2, 'beta', ROW(20, 'B'), @@ -368,7 +393,24 @@ def test_filesystem_row_complex_type(flink_gateway_url: str): MAP['c', 3], MAP['p2', ROW(200, 'silver'), 'p3', ROW(300, 'bronze')], ARRAY[TIMESTAMP '2025-12-25 00:00:00.000'], - ARRAY[ROW('z', 30)] + ARRAY[ROW('z', 30)], + MAP[ + 'env2', + ROW( + 'staging', + MAP[ + 'svc2', + ROW(TIMESTAMP '2025-06-01 12:00:00.000'), + 'svc3', + ROW(TIMESTAMP '2025-07-04 18:45:00.000') + ] + ) + ], + CAST(199.99 AS DECIMAL(10, 2)), + DATE '2025-12-25', + TIME '09:15:30', + TIMESTAMP '2025-12-25 09:15:30.456', + TO_TIMESTAMP_LTZ(1735117530456, 3) ) """, ) @@ -390,8 +432,8 @@ def test_filesystem_row_complex_type(flink_gateway_url: str): # Nested ROW-in-ROW: inner values decoded to float assert r[3]["outer_val"] == 1 - assert r[3]["inner"]["x"] == 1.5 - assert r[3]["inner"]["y"] == 2.5 + assert r[3]["inner_row"]["x"] == 1.5 + assert r[3]["inner_row"]["y"] == 2.5 # MAP assert r[4] == {"a": 1, "b": 2} @@ -404,75 +446,70 @@ def test_filesystem_row_complex_type(flink_gateway_url: str): assert r[6][0] == datetime.datetime(2024, 1, 1, 10, 0, 0) assert r[6][1] == datetime.datetime(2024, 6, 15, 14, 30, 0) - # ARRAY> + # ARRAY> arr = r[7] assert len(arr) == 2 - assert arr[0] == {"name": "x", "value": 10} - assert arr[1] == {"name": "y", "value": 20} + assert arr[0] == {"name": "x", "val": 10} + assert arr[1] == {"name": "y", "val": 20} # ── Row 2 ────────────────────────────────────────────────────── r = by_id[2] assert r[1] == "beta" assert r[2] == {"score": 20, "label": "B"} - assert r[3]["inner"]["x"] == 3.0 + assert r[3]["inner_row"]["x"] == 3.0 assert r[4] == {"c": 3} assert r[5]["p2"]["score"] == 200 assert r[5]["p3"]["level"] == "bronze" assert r[6][0] == datetime.datetime(2025, 12, 25, 0, 0, 0) - assert r[7][0] == {"name": "z", "value": 30} + assert r[7][0] == {"name": "z", "val": 30} + # ── deep_map assertions ──────────────────────────────────────── + # Row 1 + r = by_id[1] + dm = r[8] + assert dm["env1"]["label"] == "prod" + assert dm["env1"]["nested_map"]["svc1"]["ts"] == datetime.datetime( + 2024, 3, 15, 9, 30, 0 + ) -@pytest.mark.integration -def test_qqq(flink_gateway_url: str): - from flink_gateway import connect + # Row 2 + r = by_id[2] + dm = r[8] + assert dm["env2"]["label"] == "staging" + assert dm["env2"]["nested_map"]["svc2"]["ts"] == datetime.datetime( + 2025, 6, 1, 12, 0, 0 + ) + assert dm["env2"]["nested_map"]["svc3"]["ts"] == datetime.datetime( + 2025, 7, 4, 18, 45, 0 + ) + + # ── Type conversion assertions ──────────────────────────────── + # Row 1 + r = by_id[1] - with connect(flink_gateway_url) as conn: - # Create a streaming source - with conn.cursor() as cur: - cur.execute( - """ - CREATE TABLE orders - ( - id INT NOT NULL, - item STRING NOT NULL, - created_at TIMESTAMP(6) NOT NULL, - customer ROW< - first_name STRING NOT NULL, - last_name STRING NOT NULL, - age INT NOT NULL > NOT NULL, - notes STRING - ) WITH ( - 'connector' = 'datagen', - 'rows-per-second' = '5', - 'fields.id.kind' = 'sequence', - 'fields.id.start' = '1', - 'fields.id.end' = '100', - 'fields.item.length' = '12', - 'fields.customer.first_name.length' = '8', - 'fields.customer.last_name.length' = '10', - 'fields.customer.age.min' = '21', - 'fields.customer.age.max' = '65', - 'fields.notes.length' = '12' - ) - """ - ) + # DECIMAL → Decimal + assert r[9] == Decimal("99.95") + assert isinstance(r[9], Decimal) - # Query and iterate - with conn.cursor() as cur: - cur.execute( - """ - SELECT id, item, created_at, customer, notes AS note - FROM orders - """ - ) - for i, row in enumerate(cur): - id_, item, created_at, customer, note = row - print( - f"{id_}\t{item}\t{created_at.isoformat()}\t" - f"{customer['first_name']} " - f"{customer['last_name']} " - f"({customer['age']})\t" - f"{note or ''}" - ) - if i >= 4: - break + # DATE → datetime.date + assert r[10] == datetime.date(2024, 6, 15) + assert isinstance(r[10], datetime.date) + + # TIME → datetime.time + assert r[11] == datetime.time(14, 30, 0) + assert isinstance(r[11], datetime.time) + + # TIMESTAMP → datetime.datetime + assert r[12] == datetime.datetime(2024, 6, 15, 14, 30, 0, 123000) + assert isinstance(r[12], datetime.datetime) + + # TIMESTAMP_LTZ → datetime.datetime + assert isinstance(r[13], datetime.datetime) + + # Row 2 + r = by_id[2] + assert r[9] == Decimal("199.99") + assert r[10] == datetime.date(2025, 12, 25) + assert r[11] == datetime.time(9, 15, 30) + assert r[12] == datetime.datetime(2025, 12, 25, 9, 15, 30, 456000) + assert isinstance(r[13], datetime.datetime) diff --git a/uv.lock b/uv.lock index f05ef87..f272b0d 100644 --- a/uv.lock +++ b/uv.lock @@ -195,7 +195,6 @@ wheels = [ [[package]] name = "py-flink-sql-gateway" -version = "0.1.0" source = { editable = "." } dependencies = [ { name = "httpx" },