apply pyupgrade --py39-plus (#893)

* apply pyupgrade py39+

* Update media_file.py

.pre-commit-config.yaml

@@ -28,6 +28,11 @@ repos:
           - mdformat-black
           - mdformat-config
   # ----- Python formatting -----
+  - repo: https://github.com/asottile/pyupgrade
+    rev: v3.10.1
+    hooks:
+    - id: pyupgrade
+      args: [--py39-plus]
   - repo: https://github.com/psf/black
     rev: 23.7.0
     hooks:

devtools/scripts/clean_notebooks.py

@@ -2,10 +2,9 @@
 from __future__ import annotations
 
 import json
-from typing import Union
 
 
-def _clean_notebook(file: Union[str, Path]):  # pragma: no cover
+def _clean_notebook(file: str | Path):  # pragma: no cover
     """Clean ID metadata, output and execution count from jupyter notebook cells.
 
     This function overrides the existing notebook file, use with caution!

weldx/asdf/cli/welding_schema.py

@@ -3,12 +3,11 @@
 
 import sys
 from io import BytesIO
-from typing import Optional, Union
 
 
 def single_pass_weld_example(
-    out_file: Optional[Union[str, BytesIO]] = "single_pass_weld_example.asdf",
-) -> Optional[tuple[BytesIO, dict]]:
+    out_file: str | BytesIO | None = "single_pass_weld_example.asdf",
+) -> tuple[BytesIO, dict] | None:
     """Create ASDF file containing all required fields of the single_pass_weld schema.
 
     Parameters

weldx/asdf/file.py

@@ -5,10 +5,10 @@
 import io
 import pathlib
 import warnings
-from collections.abc import Iterable, Mapping, MutableMapping, Set, ValuesView
+from collections.abc import Hashable, Iterable, Mapping, MutableMapping, Set, ValuesView
 from contextlib import contextmanager
 from io import BytesIO, IOBase
-from typing import IO, Any, Dict, Hashable, Optional, Union, get_args
+from typing import IO, Any, get_args
 
 import asdf
 import numpy as np
@@ -208,16 +208,17 @@ class WeldxFile(_ProtectedViewDict):
 
     def __init__(
         self,
-        filename_or_file_like: Optional[Union[types_path_like, types_file_like]] = None,
+        filename_or_file_like: types_path_like | types_file_like | None = None,
         mode: str = "r",
         asdffile_kwargs: Mapping = None,
         write_kwargs: Mapping = None,
         tree: Mapping = None,
         sync: bool = True,
-        custom_schema: Optional[
+        custom_schema: None
+        | (
             types_path_like,
             tuple[None, types_path_like],
-        ] = None,
+        ) = None,
         software_history_entry: Mapping = None,
         compression: str = DEFAULT_ARRAY_COMPRESSION,
         copy_arrays: bool = DEFAULT_ARRAY_COPYING,
@@ -486,7 +487,7 @@ def software_history_entry(self, value: dict):
                 "version": version,
             }
         else:
-            if not isinstance(value, Dict):
+            if not isinstance(value, dict):
                 raise ValueError("expected a dictionary type")
             try:
                 test = AsdfFile(tree=dict(software=Software(value)))
@@ -519,7 +520,7 @@ def sync(
         self,
         all_array_storage: str = None,
         all_array_compression: str = "input",
-        pad_blocks: Union[float, bool] = False,
+        pad_blocks: float | bool = False,
         include_block_index: bool = True,
         version: str = None,
         **kwargs,
@@ -595,7 +596,7 @@ def get(self, key, default=None):
         """
         return super().get(key, default=default)
 
-    def update(self, mapping: Union[Mapping, Iterable] = (), **kwargs: Any):
+    def update(self, mapping: Mapping | Iterable = (), **kwargs: Any):
         """Update this file from mapping or iterable mapping and kwargs.
 
         Parameters
@@ -734,7 +735,7 @@ def asdf_library(self) -> dict:
         return self._asdf_handle["asdf_library"].copy()
 
     @property
-    def custom_schema(self) -> Optional[str, tuple[Optional[str]]]:
+    def custom_schema(self) -> str | tuple[str | None] | None:
         """Return schema used to validate the structure and types of the tree."""
         if self._schema_on_read == self._schema_on_write:
             return self._schema_on_read
@@ -752,7 +753,7 @@ def file_handle(self) -> IOBase:
 
     def copy(
         self,
-        filename_or_file_like: Optional[types_path_and_file_like] = None,
+        filename_or_file_like: types_path_and_file_like | None = None,
         overwrite: bool = False,
     ) -> WeldxFile:
         """Take a copy of this file.
@@ -834,10 +835,10 @@ def __init__(self, iterable, **kwargs):
 
     def write_to(
         self,
-        fd: Optional[types_path_and_file_like] = None,
+        fd: types_path_and_file_like | None = None,
         array_inline_threshold=None,
         **write_args,
-    ) -> Optional[types_path_and_file_like]:
+    ) -> types_path_and_file_like | None:
         """Write current contents to given file name or file type.
 
         Parameters
@@ -885,7 +886,7 @@ def header(
         self,
         use_widgets: bool = None,
         path: tuple = None,
-        _interactive: Optional[bool] = None,
+        _interactive: bool | None = None,
     ):
         """Show the header of the ASDF serialization.
 
@@ -970,7 +971,7 @@ def _fake_copy(x, _):  # take a copy of the handle to avoid side effects!
 
     def show(
         self, use_widgets=None, path=None, _interactive=None
-    ) -> Union[None, IPython.display.HTML, IPython.display.JSON]:  # noqa: F821
+    ) -> None | IPython.display.HTML | IPython.display.JSON:  # noqa: F821
         if _interactive is None:
             _interactive = is_interactive_session()
         if use_widgets is None:
@@ -995,7 +996,7 @@ def show(
     @staticmethod
     def _show_interactive(
         use_widgets: bool, buff: BytesIO, path: tuple = None
-    ) -> Union[IPython.display.HTML, IPython.display.JSON]:  # noqa: F821
+    ) -> IPython.display.HTML | IPython.display.JSON:  # noqa: F821
         from weldx.asdf.util import notebook_fileprinter
 
         if use_widgets:

weldx/asdf/types.py

@@ -2,7 +2,6 @@
 
 import functools
 import re
-from typing import Union
 
 from asdf.asdf import SerializationContext
 from asdf.extension import Converter
@@ -51,7 +50,7 @@ def from_yaml_tree_metadata(func):
     """Wrapper that will add reading metadata and userdata during form_tree methods."""
 
     @functools.wraps(func)
-    def from_yaml_tree_wrapped(self, tree: Union[dict, list, str], tag, ctx):
+    def from_yaml_tree_wrapped(self, tree: dict | list | str, tag, ctx):
         """Call default from_yaml_tree method and add metadata attributes."""
         meta_dict = {}
         if isinstance(tree, dict):  # only valid if we serialize a dict
@@ -89,7 +88,7 @@ class WeldxConverter(Converter, metaclass=WeldxConverterMeta):
     """Base class to inherit from for custom converter classes."""
 
     tags: tuple[str] = None  # note: this will be updated by WeldxConverterMeta.
-    types: tuple[Union[type, str]] = ()
+    types: tuple[type | str] = ()
 
     def to_yaml_tree(self, obj, tag: str, ctx: SerializationContext):
         raise NotImplementedError

weldx/asdf/util.py

@@ -1,11 +1,11 @@
 """Utilities for asdf files."""
 from __future__ import annotations
 
-from collections.abc import Callable, Mapping, Set
+from collections.abc import Callable, Hashable, Mapping, MutableMapping, Set
 from contextlib import contextmanager
 from io import BytesIO, TextIOBase
 from pathlib import Path
-from typing import Any, Hashable, MutableMapping, Union
+from typing import Any
 from warnings import warn
 
 import asdf
@@ -261,7 +261,7 @@ def write_read_buffer(
         return data
 
 
-def get_yaml_header(file: types_path_and_file_like, parse=False) -> Union[str, dict]:
+def get_yaml_header(file: types_path_and_file_like, parse=False) -> str | dict:
     """Read the YAML header part (excluding binary sections) of an ASDF file.
 
     Parameters
@@ -492,7 +492,7 @@ def from_yaml_tree(
     return _SerializationClass
 
 
-def get_weldx_extension(ctx: Union[SerializationContext, AsdfConfig]) -> Extension:
+def get_weldx_extension(ctx: SerializationContext | AsdfConfig) -> Extension:
     """Grab the weldx extension from list of current active extensions."""
     if isinstance(ctx, asdf.asdf.SerializationContext):
         extensions = ctx.extension_manager.extensions
@@ -508,7 +508,7 @@ def get_weldx_extension(ctx: Union[SerializationContext, AsdfConfig]) -> Extensi
     return extensions[0]
 
 
-def uri_match(patterns: Union[str, list[str]], uri: str) -> bool:
+def uri_match(patterns: str | list[str], uri: str) -> bool:
     """Returns `True` if the ASDF URI matches any of the listed patterns.
 
     See Also
@@ -521,7 +521,7 @@ def uri_match(patterns: Union[str, list[str]], uri: str) -> bool:
     return any(asdf_uri_match(p, uri) for p in patterns)
 
 
-def get_converter_for_tag(tag: str) -> Union[WeldxConverter, None]:
+def get_converter_for_tag(tag: str) -> WeldxConverter | None:
     """Get the converter class that handles a given tag."""
     converters = [s for s in WeldxConverter.__subclasses__() if uri_match(s.tags, tag)]
     if len(converters) > 1:
@@ -534,8 +534,8 @@ def get_converter_for_tag(tag: str) -> Union[WeldxConverter, None]:
 
 
 def get_highest_tag_version(
-    pattern: Union[str, list[str]], ctx: Union[SerializationContext, AsdfConfig] = None
-) -> Union[str, None]:
+    pattern: str | list[str], ctx: SerializationContext | AsdfConfig = None
+) -> str | None:
     """Get the highest available weldx extension tag version matching a pattern.
 
     Parameters
@@ -580,9 +580,7 @@ def get_highest_tag_version(
     return tags[-1]
 
 
-def _get_instance_shape(
-    instance_dict: Union[TaggedDict, dict[str, Any]]
-) -> Union[list[int], None]:
+def _get_instance_shape(instance_dict: TaggedDict | dict[str, Any]) -> list[int] | None:
     """Get the shape of an ASDF instance from its tagged dict form.
 
     Parameters
@@ -607,9 +605,7 @@ def _get_instance_shape(
     return None
 
 
-def _get_instance_units(
-    instance_dict: Union[TaggedDict, dict[str, Any]]
-) -> Union[pint.Unit, None]:
+def _get_instance_units(instance_dict: TaggedDict | dict[str, Any]) -> pint.Unit | None:
     """Get the units of an ASDF instance from its tagged dict form.
 
     Parameters
@@ -715,7 +711,7 @@ def _warn_protected_keys(self, stacklevel=3):
 
 
 def get_schema_tree(  # noqa: C901, MC0001, RUF100, codacy:ignore
-    schemafile: Union[str, Path], *, drop: set = None
+    schemafile: str | Path, *, drop: set = None
 ) -> dict:
     """Get a dictionary representation of a weldx schema file with custom formatting.
 

weldx/asdf/validators.py

@@ -4,7 +4,7 @@
 import re
 from collections import OrderedDict
 from collections.abc import Callable, Iterator, Mapping
-from typing import Any, Union
+from typing import Any
 
 from asdf.exceptions import ValidationError
 from asdf.extension import Validator
@@ -306,7 +306,7 @@ def _validate_expected_list(list_expected):
             )
 
 
-def _compare_lists(_list, list_expected) -> Union[bool, dict]:
+def _compare_lists(_list, list_expected) -> bool | dict:
     """Compare two lists.
 
     The two lists are interpreted as a list of dimensions. We compare the dimensions of
@@ -425,7 +425,7 @@ def _validate_instance_shape(
 
 def _custom_shape_validator(
     dict_test: dict[str, Any],
-    dict_expected: Union[dict[str, Any], list],
+    dict_expected: dict[str, Any] | list,
     optional: bool = False,
 ):
     """Validate dimensions which are stored in two dictionaries dict_test and

weldx/config.py

@@ -2,7 +2,6 @@
 from __future__ import annotations
 
 from pathlib import Path
-from typing import List, Union
 
 import asdf
 import pkg_resources
@@ -102,7 +101,7 @@ def name(self) -> str:
         """Get the quality standards name."""
         return self._name
 
-    def get_mappings(self, version: Union[AsdfVersion, str] = None):
+    def get_mappings(self, version: AsdfVersion | str = None):
         """Get the manifest and schema mapping for the specified version.
 
         Parameters
@@ -152,7 +151,7 @@ def add_quality_standard(standard: QualityStandard):
         Config._standards[standard.name] = standard
 
     @staticmethod
-    def enable_quality_standard(name: str, version: Union[AsdfVersion, str] = None):
+    def enable_quality_standard(name: str, version: AsdfVersion | str = None):
         """Enable a quality standard.
 
         All corresponding schemas will be used for validation during serialization and
@@ -177,7 +176,7 @@ def load_installed_standards():
         """Load all standards that are installed to the active virtual environment."""
         for entry_point in pkg_resources.iter_entry_points("weldx.standard"):
             standards = entry_point.load()()
-            if not isinstance(standards, List):
+            if not isinstance(standards, list):
                 standards = [standards]
             for standard in standards:
                 if not isinstance(standard, QualityStandard):
@@ -197,7 +196,7 @@ def add_quality_standard(standard: QualityStandard):
     Config.add_quality_standard(standard)
 
 
-def enable_quality_standard(name: str, version: Union[AsdfVersion, str] = None):
+def enable_quality_standard(name: str, version: AsdfVersion | str = None):
     """Enable a quality standard.
 
     All corresponding schemas will be used for validation during serialization and