Merge pull request #366 from epoch8/deprecate-py39

Deprecate python3.9

Author: elephantum

.github/workflows/test.yaml

@@ -26,23 +26,9 @@ jobs:
     strategy:
       matrix:
         include:
-          # - python-version: "3.8"
-          #   test-db-env: "postgres"
-          #   pip-extra: "sqlalchemy <2"
-          # - python-version: "3.8"
-          #   test-db-env: "postgres"
-          #   pip-extra: "'sqlalchemy>2'"
-          # - python-version: "3.8"
-          #   test-db-env: "sqlite"
-
-          - python-version: "3.9"
+          - python-version: "3.10"
             test-db-env: "postgres"
             pip-extra: '"sqlalchemy>2"'
-          # - python-version: "3.9"
-          #   test-db-env: "sqlite"
-
-          # - python-version: "3.10"
-          #   test-db-env: "postgres"
           # - python-version: "3.10"
           #   test-db-env: "sqlite"
 

.github/workflows/test_examples.yaml

@@ -25,8 +25,6 @@ jobs:
     strategy:
       matrix:
         python-version:
-          # - "3.8"
-          # - "3.9"
           # - "3.10"
           # - "3.11"
           - "3.12"

CHANGELOG.md

@@ -23,6 +23,8 @@ See [key-mapping.md](design-docs/2025-12-key-mapping.md) for motivation
   joining tables with different key names
 * Added `DataField` accessor for `InputSpec.keys`
 
+### Python3.9 support is deprecated
+
 ## CLI improvements:
 * Make CLI accept multiple `--name` values
 

datapipe/cli.py

@@ -1,8 +1,8 @@
 import importlib.metadata as metadata
-import os.path
+import os
 import sys
 import time
-from typing import Dict, List, Optional, cast
+from typing import cast
 
 import click
 import pandas as pd
@@ -49,7 +49,7 @@ def load_pipeline(pipeline_name: str) -> DatapipeApp:
     return app
 
 
-def parse_labels(labels: Optional[str]) -> Labels:
+def parse_labels(labels: str | None) -> Labels:
     if labels is None or labels == "":
         return []
 
@@ -68,8 +68,8 @@ def parse_labels(labels: Optional[str]) -> Labels:
 def filter_steps_by_labels_and_name(
     app: DatapipeApp,
     labels: Labels = [],
-    name_prefix: Optional[str] = None,
-) -> List[ComputeStep]:
+    name_prefix: str | None = None,
+) -> list[ComputeStep]:
     res = []
 
     name_prefixes = None if name_prefix is None else [p.strip() for p in name_prefix.split(",")]
@@ -311,7 +311,7 @@ def step(
     ctx.obj["steps"] = steps
 
 
-def to_human_repr(step: ComputeStep, extra_args: Optional[Dict] = None) -> str:
+def to_human_repr(step: ComputeStep, extra_args: dict | None = None) -> str:
     res = []
 
     res.append(f"[green][bold]{step.name}[/bold][/green] ({step.__class__.__name__})")
@@ -340,7 +340,7 @@ def to_human_repr(step: ComputeStep, extra_args: Optional[Dict] = None) -> str:
 @click.pass_context
 def step_list(ctx: click.Context, status: bool) -> None:  # noqa
     app: DatapipeApp = ctx.obj["pipeline"]
-    steps: List[ComputeStep] = ctx.obj["steps"]
+    steps: list[ComputeStep] = ctx.obj["steps"]
 
     for step in steps:
         extra_args = {}
@@ -365,7 +365,7 @@ def step_list(ctx: click.Context, status: bool) -> None:  # noqa
 @click.pass_context
 def step_run(ctx: click.Context, loop: bool, loop_delay: int) -> None:
     app: DatapipeApp = ctx.obj["pipeline"]
-    steps_to_run: List[ComputeStep] = ctx.obj["steps"]
+    steps_to_run: list[ComputeStep] = ctx.obj["steps"]
 
     executor: Executor = ctx.obj["executor"]
 
@@ -389,7 +389,7 @@ def step_run(ctx: click.Context, loop: bool, loop_delay: int) -> None:
 @click.pass_context
 def run_idx(ctx: click.Context, idx: str) -> None:
     app: DatapipeApp = ctx.obj["pipeline"]
-    steps_to_run: List[ComputeStep] = ctx.obj["steps"]
+    steps_to_run: list[ComputeStep] = ctx.obj["steps"]
     steps_to_run_names = [f"'{i.name}'" for i in steps_to_run]
     print(f"Running following steps: {', '.join(steps_to_run_names)}")
 
@@ -411,11 +411,11 @@ def run_changelist(
     start_step: str,
     loop: bool,
     loop_delay: int,
-    chunk_size: Optional[int] = None,
+    chunk_size: int | None = None,
 ) -> None:
     app: DatapipeApp = ctx.obj["pipeline"]
 
-    steps_to_run: List[ComputeStep] = ctx.obj["steps"]
+    steps_to_run: list[ComputeStep] = ctx.obj["steps"]
     if start_step is not None:
         start_step_objs = filter_steps_by_labels_and_name(app, labels=[], name_prefix=start_step)
         assert len(start_step_objs) == 1
@@ -465,7 +465,7 @@ def run_changelist(
 @click.pass_context
 def fill_metadata(ctx: click.Context) -> None:
     app: DatapipeApp = ctx.obj["pipeline"]
-    steps_to_run: List[ComputeStep] = ctx.obj["steps"]
+    steps_to_run: list[ComputeStep] = ctx.obj["steps"]
     steps_to_run_names = [f"'{i.name}'" for i in steps_to_run]
 
     for step in steps_to_run:
@@ -478,7 +478,7 @@ def fill_metadata(ctx: click.Context) -> None:
 @click.pass_context
 def reset_metadata(ctx: click.Context) -> None:  # noqa
     app: DatapipeApp = ctx.obj["pipeline"]
-    steps_to_run: List[ComputeStep] = ctx.obj["steps"]
+    steps_to_run: list[ComputeStep] = ctx.obj["steps"]
     steps_to_run_names = [f"'{i.name}'" for i in steps_to_run]
     print(f"Resetting following steps: {', '.join(steps_to_run_names)}")
 

datapipe/compute.py

@@ -2,7 +2,7 @@
 import logging
 from abc import ABC, abstractmethod
 from dataclasses import dataclass
-from typing import Dict, Iterable, List, Literal, Optional, Sequence, Tuple, Union
+from typing import Iterable, Literal, Sequence
 
 from opentelemetry import trace
 from sqlalchemy import Column
@@ -21,13 +21,13 @@
 @dataclass
 class Table:
     store: TableStore
-    name: Optional[str] = None
+    name: str | None = None
 
 
 class Catalog:
-    def __init__(self, catalog: Dict[str, Table]):
+    def __init__(self, catalog: dict[str, Table]):
         self.catalog = catalog
-        self.data_tables: Dict[str, DataTable] = {}
+        self.data_tables: dict[str, DataTable] = {}
 
     def add_datatable(self, name: str, dt: Table):
         self.catalog[name] = dt
@@ -94,10 +94,10 @@ class ComputeInput:
     #
     # Example: {"idx_col": "meta_col"} means that to get idx_col value
     # we should read meta_col from meta table
-    keys: Optional[Dict[str, FieldAccessor]] = None
+    keys: dict[str, FieldAccessor] | None = None
 
     @property
-    def primary_keys(self) -> List[str]:
+    def primary_keys(self) -> list[str]:
         if self.keys:
             return list(self.keys.keys())
         else:
@@ -152,10 +152,10 @@ class ComputeStep:
     def __init__(
         self,
         name: str,
-        input_dts: Sequence[Union[ComputeInput, DataTable]],
-        output_dts: List[DataTable],
-        labels: Optional[Labels] = None,
-        executor_config: Optional[ExecutorConfig] = None,
+        input_dts: Sequence[ComputeInput | DataTable],
+        output_dts: list[DataTable],
+        labels: Labels | None = None,
+        executor_config: ExecutorConfig | None = None,
     ) -> None:
         self._name = name
         # Нормализация input_dts: автоматически оборачиваем DataTable в ComputeInput
@@ -222,41 +222,41 @@ def validate(self) -> None:
     def get_full_process_ids(
         self,
         ds: DataStore,
-        chunk_size: Optional[int] = None,
-        run_config: Optional[RunConfig] = None,
-    ) -> Tuple[int, Iterable[IndexDF]]:
+        chunk_size: int | None = None,
+        run_config: RunConfig | None = None,
+    ) -> tuple[int, Iterable[IndexDF]]:
         raise NotImplementedError()
 
     def get_change_list_process_ids(
         self,
         ds: DataStore,
         change_list: ChangeList,
-        run_config: Optional[RunConfig] = None,
-    ) -> Tuple[int, Iterable[IndexDF]]:
+        run_config: RunConfig | None = None,
+    ) -> tuple[int, Iterable[IndexDF]]:
         raise NotImplementedError()
 
     def run_full(
         self,
         ds: DataStore,
-        run_config: Optional[RunConfig] = None,
-        executor: Optional[Executor] = None,
+        run_config: RunConfig | None = None,
+        executor: Executor | None = None,
     ) -> None:
         raise NotImplementedError()
 
     def run_changelist(
         self,
         ds: DataStore,
         change_list: ChangeList,
-        run_config: Optional[RunConfig] = None,
-        executor: Optional[Executor] = None,
+        run_config: RunConfig | None = None,
+        executor: Executor | None = None,
     ) -> ChangeList:
         raise NotImplementedError()
 
     def run_idx(
         self,
         ds: DataStore,
         idx: IndexDF,
-        run_config: Optional[RunConfig] = None,
+        run_config: RunConfig | None = None,
     ) -> ChangeList:
         raise NotImplementedError()
 
@@ -270,7 +270,7 @@ class PipelineStep(ABC):
     """
 
     @abstractmethod
-    def build_compute(self, ds: DataStore, catalog: Catalog) -> List[ComputeStep]:
+    def build_compute(self, ds: DataStore, catalog: Catalog) -> list[ComputeStep]:
         raise NotImplementedError
 
 
@@ -288,11 +288,11 @@ def __init__(self, ds: DataStore, catalog: Catalog, pipeline: Pipeline):
         self.steps = build_compute(ds, catalog, pipeline)
 
 
-def build_compute(ds: DataStore, catalog: Catalog, pipeline: Pipeline) -> List[ComputeStep]:
+def build_compute(ds: DataStore, catalog: Catalog, pipeline: Pipeline) -> list[ComputeStep]:
     with tracer.start_as_current_span("build_compute"):
         catalog.init_all_tables(ds)
 
-        compute_pipeline: List[ComputeStep] = []
+        compute_pipeline: list[ComputeStep] = []
 
         for step in pipeline.steps:
             compute_pipeline.extend(step.build_compute(ds, catalog))
@@ -304,7 +304,7 @@ def build_compute(ds: DataStore, catalog: Catalog, pipeline: Pipeline) -> List[C
         return compute_pipeline
 
 
-def print_compute(steps: List[ComputeStep]) -> None:
+def print_compute(steps: list[ComputeStep]) -> None:
     import pprint
 
     pprint.pp(steps)
@@ -313,8 +313,8 @@ def print_compute(steps: List[ComputeStep]) -> None:
 def run_steps(
     ds: DataStore,
     steps: Sequence[ComputeStep],
-    run_config: Optional[RunConfig] = None,
-    executor: Optional[Executor] = None,
+    run_config: RunConfig | None = None,
+    executor: Executor | None = None,
 ) -> None:
     for step in steps:
         with tracer.start_as_current_span(
@@ -332,7 +332,7 @@ def run_pipeline(
     ds: DataStore,
     catalog: Catalog,
     pipeline: Pipeline,
-    run_config: Optional[RunConfig] = None,
+    run_config: RunConfig | None = None,
 ) -> None:
     steps = build_compute(ds, catalog, pipeline)
     run_steps(ds, steps, run_config)
@@ -343,7 +343,7 @@ def run_changelist(
     catalog: Catalog,
     pipeline: Pipeline,
     changelist: ChangeList,
-    run_config: Optional[RunConfig] = None,
+    run_config: RunConfig | None = None,
 ) -> None:
     steps = build_compute(ds, catalog, pipeline)
 
@@ -352,10 +352,10 @@ def run_changelist(
 
 def run_steps_changelist(
     ds: DataStore,
-    steps: List[ComputeStep],
+    steps: list[ComputeStep],
     changelist: ChangeList,
-    run_config: Optional[RunConfig] = None,
-    executor: Optional[Executor] = None,
+    run_config: RunConfig | None = None,
+    executor: Executor | None = None,
 ) -> None:
     # FIXME extract Batch* steps to separate module
     from datapipe.step.batch_transform import BaseBatchTransformStep

datapipe/datatable.py

@@ -1,5 +1,5 @@
 import logging
-from typing import TYPE_CHECKING, Any, Dict, Optional, cast
+from typing import TYPE_CHECKING, Any, cast
 
 import pandas as pd
 from opentelemetry import trace
@@ -38,10 +38,10 @@ def __init__(
         self.primary_schema = meta.primary_schema
         self.primary_keys = meta.primary_keys
 
-    def get_metadata(self, idx: Optional[IndexDF] = None) -> MetadataDF:
+    def get_metadata(self, idx: IndexDF | None = None) -> MetadataDF:
         return self.meta.get_metadata(idx)
 
-    def get_data(self, idx: Optional[IndexDF] = None) -> DataDF:
+    def get_data(self, idx: IndexDF | None = None) -> DataDF:
         return self.table_store.read_rows(self.meta.get_existing_idx(idx))
 
     def reset_metadata(self):
@@ -56,9 +56,9 @@ def get_size(self) -> int:
     def store_chunk(
         self,
         data_df: DataDF,
-        processed_idx: Optional[IndexDF] = None,
-        now: Optional[float] = None,
-        run_config: Optional[RunConfig] = None,
+        processed_idx: IndexDF | None = None,
+        now: float | None = None,
+        run_config: RunConfig | None = None,
     ) -> IndexDF:
         """
         Записать новые данные в таблицу.
@@ -134,8 +134,8 @@ def store_chunk(
     def delete_by_idx(
         self,
         idx: IndexDF,
-        now: Optional[float] = None,
-        run_config: Optional[RunConfig] = None,
+        now: float | None = None,
+        run_config: RunConfig | None = None,
     ) -> None:
         if len(idx) > 0:
             logger.debug(f"Deleting {len(idx.index)} rows from {self.name} data")
@@ -146,8 +146,8 @@ def delete_by_idx(
     def delete_stale_by_process_ts(
         self,
         process_ts: float,
-        now: Optional[float] = None,
-        run_config: Optional[RunConfig] = None,
+        now: float | None = None,
+        run_config: RunConfig | None = None,
     ) -> None:
         for deleted_df in self.meta.get_stale_idx(process_ts, run_config=run_config):
             deleted_idx = data_to_index(deleted_df, self.primary_keys)
@@ -165,7 +165,7 @@ def __init__(
 
         self.meta_dbconn = meta_dbconn
         self.event_logger = EventLogger()
-        self.tables: Dict[str, DataTable] = {}
+        self.tables: dict[str, DataTable] = {}
 
         # TODO move initialization outside
         self.meta_plane = SQLMetaPlane(dbconn=meta_dbconn, create_meta_table=create_meta_table)