Use type union operator in 34 files

- Remove use of typing.{Optional,Union}.
- Import generics from collections.abc.

Author: khaeru

sdmx/client.py

@@ -1,6 +1,6 @@
 import logging
 from functools import partial
-from typing import IO, TYPE_CHECKING, Any, Optional, Union
+from typing import IO, TYPE_CHECKING, Any
 from warnings import warn
 
 import requests
@@ -74,7 +74,7 @@ def __init__(
         self,
         source=None,
         *,
-        session: Optional["requests.Session"] = None,
+        session: "requests.Session" | None = None,
         log_level=None,
         **session_opts,
     ):
@@ -333,9 +333,9 @@ def _collect(*keywords):
 
     def get(
         self,
-        resource_type: Union[str, Resource, None] = None,
-        resource_id: Optional[str] = None,
-        tofile: Union["os.PathLike", IO, None] = None,
+        resource_type: str | Resource | None = None,
+        resource_id: str | None = None,
+        tofile: "os.PathLike" | IO | None = None,
         use_cache: bool = False,
         dry_run: bool = False,
         **kwargs,

sdmx/compare.py

@@ -9,7 +9,7 @@
 from copy import copy
 from dataclasses import dataclass, fields, is_dataclass
 from functools import singledispatch
-from typing import Any, TypeVar, Union
+from typing import Any, TypeVar
 
 import lxml.etree
 
@@ -157,7 +157,7 @@ def compare_dataclass(left, right, opts: Options, context: str) -> bool:
 @compare.register(int)
 @compare.register(str)
 @compare.register(datetime.date)
-def _eq(left: Union[int, str, datetime.date], right, opts, context=""):
+def _eq(left: int | str | datetime.date, right, opts, context=""):
     """Built-in types that must compare equal."""
     return left == right or (not opts.strict and right is None)
 
@@ -168,7 +168,7 @@ def _eq(left: Union[int, str, datetime.date], right, opts, context=""):
 @compare.register(float)
 @compare.register(type)
 @compare.register(enum.Enum)
-def _is(left: Union[None, bool, float, type, enum.Enum], right, opts, context):
+def _is(left: None | bool | float | type | enum.Enum, right, opts, context):
     """Built-in types that must compare identical."""
     return left is right or (not opts.strict and right is None or left is None)
 
@@ -203,7 +203,7 @@ def _(left: dict, right, opts, context=""):
 # TODO When dropping support for Python <=3.10, change to '@compare.register'
 @compare.register(list)
 @compare.register(set)
-def _(left: Union[list, set], right, opts, context=""):
+def _(left: list | set, right, opts, context=""):
     if len(left) != len(right):
         opts.log(f"Mismatched length: {len(left)} != {len(right)}")
         return False

sdmx/convert/pandas.py

@@ -5,7 +5,7 @@
 from dataclasses import InitVar, dataclass, field
 from itertools import chain, product, repeat
 from types import SimpleNamespace
-from typing import TYPE_CHECKING, Any, Optional, Union, cast
+from typing import TYPE_CHECKING, Any, cast
 from warnings import warn
 
 import numpy as np
@@ -34,7 +34,7 @@
     class ToDatetimeKeywords(TypedDict, total=False):
         format: str
 
-    KeyOrAttributeValue = Union["common.KeyValue", "common.AttributeValue"]
+    KeyOrAttributeValue = common.KeyValue | common.AttributeValue
 
 
 _HAS_PANDAS_2 = pd.__version__.split(".")[0] >= "2"
@@ -150,8 +150,8 @@ class ColumnSpec:
 
     def __init__(
         self,
-        pc: Optional["PandasConverter"] = None,
-        ds: Optional["common.BaseDataSet"] = None,
+        pc: "PandasConverter" | None = None,
+        ds: "common.BaseDataSet" | None = None,
     ) -> None:
         if pc is None or ds is None:
             return  # Empty/placeholder
@@ -227,9 +227,9 @@ def __init__(
 
     @staticmethod
     def _maybe_construct_dsd(
-        dsd: Optional["common.BaseDataStructureDefinition"],
+        dsd: "common.BaseDataStructureDefinition" | None,
         obs: "common.BaseObservation",
-    ) -> Union["v21.DataStructureDefinition", "v30.DataStructure"]:
+    ) -> "v21.DataStructureDefinition" | "v30.DataStructure":
         """If `dsd` is None, construct a DSD by inspection of `obs`."""
         if dsd is not None:
             return dsd
@@ -275,7 +275,7 @@ def convert_obs(self, obs: "common.BaseObservation") -> list:
         key = obs.key
         if self.constraint and key not in self.constraint:
             # Emit an empty row to be dropped
-            result: Iterable[Union[str, None]] = repeat(None, len(self.obs))
+            result: Iterable[str | None] = repeat(None, len(self.obs))
         else:
             # Observation values
             # FIXME Handled CodedObservationValue, similar to AttributeValue
@@ -288,7 +288,7 @@ def convert_obs(self, obs: "common.BaseObservation") -> list:
             self.add_obs_attrib(avs)
 
             # - Convert the observation Key using key Columns.
-            # - Convert the value to Optional[str].
+            # - Convert the value to str | None.
             # - Convert the attribute values using attribute Columns.
             result = chain(
                 [c(key.values) for c in self.key],
@@ -314,26 +314,26 @@ class PandasConverter(DispatchConverter):
     attributes: Attributes = Attributes.none
 
     #: If given, only Observations included by the *constraint* are returned.
-    constraint: Optional["ContentConstraint"] = None
+    constraint: "ContentConstraint | None" = None
 
     #: Datatype for observation values. If :any:`None`, data values remain
     #: :class:`object`/:class:`str`.
-    dtype: Union[type["np.generic"], type["ExtensionDtype"], str, None] = np.float64
+    dtype: type["np.generic"] | type["ExtensionDtype"] | str | None = np.float64
 
     #: Axis on which to place a time dimension. One of:
     #:
     #: - :py:`-1`: disabled.
     #: - :py:`0, "index"`: first/index axis.
     #: - :py:`1, "columns"`: second/columns axis.
-    datetime_axis: Union[int, str] = -1
+    datetime_axis: int | str = -1
 
     #: Dimension to convert to :class:`pandas.DatetimeIndex`. A :class:`str` value is
     #: interpreted as a dimension ID.
-    datetime_dimension: Optional["common.DimensionComponent"] = None
+    datetime_dimension: "common.DimensionComponent | None" = None
 
     #: Frequency for conversion to :class:`pandas.PeriodIndex`. A :class:`str` value is
     #: interpreted as one of the :ref:`pd:timeseries.period_aliases`.
-    datetime_freq: Optional["DateOffset"] = None
+    datetime_freq: "DateOffset | None" = None
 
     #: include : iterable of str or str, optional
     #:     One or more of the attributes of the StructureMessage ('category_scheme',
@@ -367,9 +367,7 @@ class PandasConverter(DispatchConverter):
     # Columns to be set as index levels, then unstacked.
     _unstack: list[str] = field(default_factory=list)
 
-    _context: dict[Union[str, type], Any] = field(
-        default_factory=lambda: dict(compat=False)
-    )
+    _context: dict[str | type, Any] = field(default_factory=lambda: dict(compat=False))
 
     def get_components(self, kind) -> list["common.Component"]:
         """Return an appropriate list of dimensions or attributes."""
@@ -448,7 +446,7 @@ def handle_datetime(self, value: Any) -> None:
         else:
             raise TypeError(f"PandasConverter(…, datetime={type(value)})")
 
-    def __post_init__(self, datetime: Any, rtype: Optional[str]) -> None:
+    def __post_init__(self, datetime: Any, rtype: str | None) -> None:
         """Transform and validate arguments."""
         # Raise on unsupported arguments
         if isinstance(
@@ -626,7 +624,7 @@ def convert_structuremessage(c: "PandasConverter", obj: message.StructureMessage
         Keys are StructureMessage attributes; values are pandas objects.
     """
     attrs = sorted(c.include)
-    result: DictLike[str, Union[pd.Series, pd.DataFrame]] = DictLike()
+    result: DictLike[str, pd.Series | pd.DataFrame] = DictLike()
     for a in attrs:
         dl = c.convert(getattr(obj, a))
         if len(dl):
@@ -755,17 +753,15 @@ def _convert_datetime(df: "pd.DataFrame", c: "PandasConverter") -> "pd.DataFrame
     return df.assign(**{dim.id: pd.to_datetime(df[dim.id], **dt_kw)})
 
 
-def _ensure_multiindex(obj: Union[pd.Series, pd.DataFrame]):
+def _ensure_multiindex(obj: pd.Series | pd.DataFrame):
     if not isinstance(obj.index, pd.MultiIndex):
         obj.index = pd.MultiIndex.from_product(
             [obj.index.to_list()], names=[obj.index.name]
         )
     return obj
 
 
-def _reshape(
-    df: "pd.DataFrame", c: "PandasConverter"
-) -> Union[pd.Series, pd.DataFrame]:
+def _reshape(df: "pd.DataFrame", c: "PandasConverter") -> pd.Series | pd.DataFrame:
     """Reshape `df` to provide expected return types."""
 
     if c._strict:
@@ -790,7 +786,7 @@ def _reshape(
     return result
 
 
-def _to_periodindex(obj: Union["pd.Series", "pd.DataFrame"], c: "PandasConverter"):
+def _to_periodindex(obj: "pd.Series" | "pd.DataFrame", c: "PandasConverter"):
     """Convert a 1-D datetime index on `obj` to a PeriodIndex."""
     result = obj
 
@@ -887,7 +883,7 @@ def add_item(item):
         add_item(item)
 
     # Convert to DataFrame
-    result: Union[pd.DataFrame, pd.Series] = pd.DataFrame.from_dict(
+    result: pd.DataFrame | pd.Series = pd.DataFrame.from_dict(
         items,
         orient="index",
         dtype=object,  # type: ignore [arg-type]

sdmx/dictlike.py

@@ -1,7 +1,7 @@
 import logging
-import typing
+from collections.abc import MutableMapping
 from dataclasses import fields
-from typing import Generic, TypeVar, Union, get_args, get_origin
+from typing import Generic, TypeVar, get_args, get_origin
 
 from sdmx.compare import Comparable
 
@@ -11,7 +11,7 @@
 VT = TypeVar("VT")
 
 
-class DictLike(dict, typing.MutableMapping[KT, VT], Comparable):
+class DictLike(dict, MutableMapping[KT, VT], Comparable):
     """Container with features of :class:`dict`, attribute access, and validation."""
 
     __slots__ = ("__dict__", "_types")
@@ -32,7 +32,7 @@ def with_types(cls, key_type, value_type):
         result._types = (key_type, value_type)
         return result
 
-    def __getitem__(self, key: Union[KT, int]) -> VT:
+    def __getitem__(self, key: KT | int) -> VT:
         """:meth:`dict.__getitem__` with integer access."""
         try:
             return super().__getitem__(key)

sdmx/experimental.py

@@ -13,7 +13,7 @@
 
 """
 
-from typing import Optional, Text
+from typing import Text
 
 import pandas as pd
 
@@ -31,10 +31,10 @@
 
 class DataSet(AnnotableArtefact):
     # SDMX-IM features
-    action: Optional[ActionType] = None
+    action: ActionType | None = None
     attrib: DictLike[str, AttributeValue] = DictLike()
-    valid_from: Optional[Text] = None
-    structured_by: Optional[DataStructureDefinition] = None
+    valid_from: Text | None = None
+    structured_by: DataStructureDefinition | None = None
 
     # Internal storage: a pd.DataFrame with columns:
     # - 'value': the Observation value.

sdmx/format/__init__.py

@@ -2,7 +2,7 @@
 from dataclasses import InitVar, dataclass, field
 from enum import Enum, IntFlag
 from functools import lru_cache
-from typing import Literal, Optional, Union
+from typing import Literal
 
 from sdmx.util import parse_content_type
 
@@ -48,12 +48,12 @@ class MediaType:
 
     #: Format version.
     version: Version = field(init=False)
-    _version: InitVar[Union[str, Version]]
+    _version: InitVar[str | Version]
 
     flags: Flag = Flag(0)
 
     #: Specify the full media type string.
-    full: Optional[str] = None
+    full: str | None = None
 
     def __post_init__(self, _version):
         self.__dict__["version"] = Version[_version]

sdmx/format/xml/common.py

@@ -1,12 +1,13 @@
 import logging
 import re
 import zipfile
+from collections.abc import Iterable, Mapping
 from functools import lru_cache
 from itertools import chain
 from operator import itemgetter
 from pathlib import Path
 from shutil import copytree
-from typing import IO, Iterable, Mapping, Optional, Union, cast
+from typing import IO, cast
 
 from lxml import etree
 from lxml.etree import QName
@@ -98,9 +99,9 @@
 
 
 def validate_xml(
-    msg: Union[Path, IO],
-    schema_dir: Optional[Path] = None,
-    version: Union[str, Version] = Version["2.1"],
+    msg: Path | IO,
+    schema_dir: Path | None = None,
+    version: str | Version = Version["2.1"],
     max_errors: int = -1,
 ) -> bool:
     """Validate SDMX-ML in `msg` against the XML Schema (XSD) documents.
@@ -160,8 +161,8 @@ def validate_xml(
 
 
 def construct_schema(
-    schema_dir: Optional[Path] = None,
-    version: Union[str, Version] = Version["2.1"],
+    schema_dir: Path | None = None,
+    version: str | Version = Version["2.1"],
 ) -> "etree.XMLSchema":
     """Construct a :class:`lxml.etree.XMLSchema` for SDMX-ML of the given `version`.
 
@@ -279,7 +280,7 @@ def _extracted_zipball(version: Version, force: bool = False) -> Path:
 
 
 def _handle_validate_args(
-    schema_dir: Optional[Path], version: Union[str, Version]
+    schema_dir: Path | None, version: str | Version
 ) -> tuple[Path, Version]:
     """Handle arguments for :func:`.install_schemas` and :func:`.validate_xml`."""
     import platformdirs
@@ -294,16 +295,16 @@ def _handle_validate_args(
         ) from None
 
     # If the user has no preference, download the schemas to the local cache directory
-    if not schema_dir:
+    if schema_dir is None:
         schema_dir = platformdirs.user_cache_path("sdmx") / version.name
     schema_dir.mkdir(exist_ok=True, parents=True)
 
     return schema_dir, version
 
 
 def install_schemas(
-    schema_dir: Optional[Path] = None,
-    version: Union[str, Version] = Version["2.1"],
+    schema_dir: Path | None = None,
+    version: str | Version = Version["2.1"],
 ) -> Path:
     """Install SDMX-ML XML Schema documents for use with :func:`.validate_xml`.
 
@@ -334,7 +335,7 @@ def install_schemas(
 class XMLFormat(Format):
     """Information about an SDMX-ML format."""
 
-    NS: Mapping[str, Optional[str]]
+    NS: Mapping[str, str | None]
     _class_tag: list
 
     def __init__(self, model, base_ns: str, class_tag: Iterable[tuple[str, str]]):
@@ -370,7 +371,7 @@ def ns_prefix(self, url) -> str:
     _NS_PATTERN = re.compile(r"(\{(?P<ns>.*)\}|(?P<ns_prefix>.*):)?(?P<localname>.*)")
 
     @lru_cache()
-    def qname(self, ns_or_name: str, name: Optional[str] = None) -> QName:
+    def qname(self, ns_or_name: str, name: str | None = None) -> QName:
         """Return a fully-qualified tag `name` in namespace `ns`."""
         if isinstance(ns_or_name, QName):
             # Already a QName; do nothing
@@ -395,7 +396,7 @@ def qname(self, ns_or_name: str, name: Optional[str] = None) -> QName:
         return QName(ns, name)
 
     @lru_cache()
-    def class_for_tag(self, tag) -> Optional[type]:
+    def class_for_tag(self, tag) -> type | None:
         """Return a message or model class for an XML tag."""
         qname = self.qname(tag)
         results = map(itemgetter(0), filter(lambda ct: ct[1] == qname, self._class_tag))