From 49107f57bb94e563b87101e4915615525bd43432 Mon Sep 17 00:00:00 2001 From: Nejc Habjan Date: Wed, 5 Feb 2025 19:13:02 +0100 Subject: [PATCH] refactor: use more python3.9 syntax --- docs/conf.py | 1 - gitlab/__init__.py | 1 - gitlab/_backends/protocol.py | 22 +- gitlab/_backends/requests_backend.py | 24 +-- gitlab/base.py | 70 ++++--- gitlab/cli.py | 29 ++- gitlab/client.py | 200 +++++++++---------- gitlab/config.py | 33 +-- gitlab/exceptions.py | 12 +- gitlab/mixins.py | 190 +++++++++--------- gitlab/types.py | 26 +-- gitlab/utils.py | 56 +++--- gitlab/v4/cli.py | 63 +++--- gitlab/v4/objects/appearance.py | 10 +- gitlab/v4/objects/artifacts.py | 16 +- gitlab/v4/objects/bulk_imports.py | 4 +- gitlab/v4/objects/clusters.py | 10 +- gitlab/v4/objects/commits.py | 30 ++- gitlab/v4/objects/container_registry.py | 4 +- gitlab/v4/objects/deploy_keys.py | 8 +- gitlab/v4/objects/deployments.py | 10 +- gitlab/v4/objects/environments.py | 6 +- gitlab/v4/objects/epics.py | 10 +- gitlab/v4/objects/features.py | 14 +- gitlab/v4/objects/files.py | 37 ++-- gitlab/v4/objects/groups.py | 28 +-- gitlab/v4/objects/invitations.py | 10 +- gitlab/v4/objects/issues.py | 22 +- gitlab/v4/objects/job_token_scope.py | 6 +- gitlab/v4/objects/jobs.py | 27 ++- gitlab/v4/objects/keys.py | 6 +- gitlab/v4/objects/labels.py | 22 +- gitlab/v4/objects/ldap.py | 6 +- gitlab/v4/objects/members.py | 4 +- gitlab/v4/objects/merge_request_approvals.py | 10 +- gitlab/v4/objects/merge_requests.py | 36 ++-- gitlab/v4/objects/packages.py | 22 +- gitlab/v4/objects/pipelines.py | 24 ++- gitlab/v4/objects/projects.py | 76 ++++--- gitlab/v4/objects/releases.py | 4 +- gitlab/v4/objects/repositories.py | 46 ++--- gitlab/v4/objects/resource_groups.py | 4 +- gitlab/v4/objects/runners.py | 6 +- gitlab/v4/objects/secure_files.py | 10 +- gitlab/v4/objects/settings.py | 10 +- gitlab/v4/objects/sidekiq.py | 16 +- gitlab/v4/objects/snippets.py | 23 +-- gitlab/v4/objects/topics.py | 11 +- gitlab/v4/objects/users.py | 48 ++--- tests/functional/api/test_packages.py | 6 +- tests/functional/api/test_projects.py | 2 +- tests/functional/api/test_repository.py | 4 - tests/functional/conftest.py | 8 +- tests/functional/helpers.py | 6 +- tests/unit/base/test_rest_object.py | 8 +- tests/unit/helpers.py | 7 +- tests/unit/meta/test_imports.py | 2 +- tests/unit/meta/test_mro.py | 8 +- tests/unit/test_cli.py | 3 - 59 files changed, 698 insertions(+), 719 deletions(-) diff --git a/docs/conf.py b/docs/conf.py index fadf2b6a9..32e11abb9 100644 --- a/docs/conf.py +++ b/docs/conf.py @@ -1,5 +1,4 @@ #!/usr/bin/env python3 -# -*- coding: utf-8 -*- # # python-gitlab documentation build configuration file, created by # sphinx-quickstart on Mon Dec 8 15:17:39 2014. diff --git a/gitlab/__init__.py b/gitlab/__init__.py index 17a6052b5..e7a24cb1d 100644 --- a/gitlab/__init__.py +++ b/gitlab/__init__.py @@ -1,4 +1,3 @@ -# -*- coding: utf-8 -*- # # Copyright (C) 2013-2019 Gauvain Pocentek, 2019-2023 python-gitlab team # diff --git a/gitlab/_backends/protocol.py b/gitlab/_backends/protocol.py index 72cee226d..05721bc77 100644 --- a/gitlab/_backends/protocol.py +++ b/gitlab/_backends/protocol.py @@ -1,15 +1,11 @@ +from __future__ import annotations + import abc -import sys -from typing import Any, Dict, Optional, Union +from typing import Any, Protocol import requests from requests_toolbelt.multipart.encoder import MultipartEncoder # type: ignore -if sys.version_info >= (3, 8): - from typing import Protocol -else: - from typing_extensions import Protocol - class BackendResponse(Protocol): @abc.abstractmethod @@ -22,11 +18,11 @@ def http_request( self, method: str, url: str, - json: Optional[Union[Dict[str, Any], bytes]], - data: Optional[Union[Dict[str, Any], MultipartEncoder]], - params: Optional[Any], - timeout: Optional[float], - verify: Optional[Union[bool, str]], - stream: Optional[bool], + json: dict[str, Any] | bytes | None, + data: dict[str, Any] | MultipartEncoder | None, + params: Any | None, + timeout: float | None, + verify: bool | str | None, + stream: bool | None, **kwargs: Any, ) -> BackendResponse: ... diff --git a/gitlab/_backends/requests_backend.py b/gitlab/_backends/requests_backend.py index 79e3cbf12..32b45ad9b 100644 --- a/gitlab/_backends/requests_backend.py +++ b/gitlab/_backends/requests_backend.py @@ -1,7 +1,7 @@ from __future__ import annotations import dataclasses -from typing import Any, BinaryIO, Dict, Optional, TYPE_CHECKING, Union +from typing import Any, BinaryIO, TYPE_CHECKING import requests from requests import PreparedRequest @@ -44,8 +44,8 @@ def __call__(self, r: PreparedRequest) -> PreparedRequest: @dataclasses.dataclass class SendData: content_type: str - data: Optional[Union[Dict[str, Any], MultipartEncoder]] = None - json: Optional[Union[Dict[str, Any], bytes]] = None + data: dict[str, Any] | MultipartEncoder | None = None + json: dict[str, Any] | bytes | None = None def __post_init__(self) -> None: if self.json is not None and self.data is not None: @@ -84,7 +84,7 @@ def json(self) -> Any: class RequestsBackend(protocol.Backend): - def __init__(self, session: Optional[requests.Session] = None) -> None: + def __init__(self, session: requests.Session | None = None) -> None: self._client: requests.Session = session or requests.Session() @property @@ -93,8 +93,8 @@ def client(self) -> requests.Session: @staticmethod def prepare_send_data( - files: Optional[Dict[str, Any]] = None, - post_data: Optional[Union[Dict[str, Any], bytes, BinaryIO]] = None, + files: dict[str, Any] | None = None, + post_data: dict[str, Any] | bytes | BinaryIO | None = None, raw: bool = False, ) -> SendData: if files: @@ -130,12 +130,12 @@ def http_request( self, method: str, url: str, - json: Optional[Union[Dict[str, Any], bytes]] = None, - data: Optional[Union[Dict[str, Any], MultipartEncoder]] = None, - params: Optional[Any] = None, - timeout: Optional[float] = None, - verify: Optional[Union[bool, str]] = True, - stream: Optional[bool] = False, + json: dict[str, Any] | bytes | None = None, + data: dict[str, Any] | MultipartEncoder | None = None, + params: Any | None = None, + timeout: float | None = None, + verify: bool | str | None = True, + stream: bool | None = False, **kwargs: Any, ) -> RequestsResponse: """Make HTTP request diff --git a/gitlab/base.py b/gitlab/base.py index 9a8c80acf..7637df881 100644 --- a/gitlab/base.py +++ b/gitlab/base.py @@ -1,20 +1,18 @@ +from __future__ import annotations + import copy import importlib import json import pprint import textwrap +from collections.abc import Iterable from types import ModuleType from typing import ( Any, ClassVar, - Dict, Generic, - Iterable, - Optional, - Type, TYPE_CHECKING, TypeVar, - Union, ) import gitlab @@ -51,20 +49,20 @@ class RESTObject: object's ``__repr__()`` method. """ - _id_attr: Optional[str] = "id" - _attrs: Dict[str, Any] + _id_attr: str | None = "id" + _attrs: dict[str, Any] _created_from_list: bool # Indicates if object was created from a list() action _module: ModuleType - _parent_attrs: Dict[str, Any] - _repr_attr: Optional[str] = None - _updated_attrs: Dict[str, Any] + _parent_attrs: dict[str, Any] + _repr_attr: str | None = None + _updated_attrs: dict[str, Any] _lazy: bool - manager: "RESTManager[Any]" + manager: RESTManager[Any] def __init__( self, - manager: "RESTManager[Any]", - attrs: Dict[str, Any], + manager: RESTManager[Any], + attrs: dict[str, Any], *, created_from_list: bool = False, lazy: bool = False, @@ -88,13 +86,13 @@ def __init__( self.__dict__["_parent_attrs"] = self.manager.parent_attrs self._create_managers() - def __getstate__(self) -> Dict[str, Any]: + def __getstate__(self) -> dict[str, Any]: state = self.__dict__.copy() module = state.pop("_module") state["_module_name"] = module.__name__ return state - def __setstate__(self, state: Dict[str, Any]) -> None: + def __setstate__(self, state: dict[str, Any]) -> None: module_name = state.pop("_module_name") self.__dict__.update(state) self.__dict__["_module"] = importlib.import_module(module_name) @@ -147,7 +145,7 @@ def __getattr__(self, name: str) -> Any: def __setattr__(self, name: str, value: Any) -> None: self.__dict__["_updated_attrs"][name] = value - def asdict(self, *, with_parent_attrs: bool = False) -> Dict[str, Any]: + def asdict(self, *, with_parent_attrs: bool = False) -> dict[str, Any]: data = {} if with_parent_attrs: data.update(copy.deepcopy(self._parent_attrs)) @@ -156,7 +154,7 @@ def asdict(self, *, with_parent_attrs: bool = False) -> Dict[str, Any]: return data @property - def attributes(self) -> Dict[str, Any]: + def attributes(self) -> dict[str, Any]: return self.asdict(with_parent_attrs=True) def to_json(self, *, with_parent_attrs: bool = False, **kwargs: Any) -> str: @@ -231,11 +229,11 @@ def _create_managers(self) -> None: # Since we have our own __setattr__ method, we can't use setattr() self.__dict__[attr] = manager - def _update_attrs(self, new_attrs: Dict[str, Any]) -> None: + def _update_attrs(self, new_attrs: dict[str, Any]) -> None: self.__dict__["_updated_attrs"] = {} self.__dict__["_attrs"] = new_attrs - def get_id(self) -> Optional[Union[int, str]]: + def get_id(self) -> int | str | None: """Returns the id of the resource.""" if self._id_attr is None or not hasattr(self, self._id_attr): return None @@ -245,7 +243,7 @@ def get_id(self) -> Optional[Union[int, str]]: return id_val @property - def _repr_value(self) -> Optional[str]: + def _repr_value(self) -> str | None: """Safely returns the human-readable resource name if present.""" if self._repr_attr is None or not hasattr(self, self._repr_attr): return None @@ -255,7 +253,7 @@ def _repr_value(self) -> Optional[str]: return repr_val @property - def encoded_id(self) -> Optional[Union[int, str]]: + def encoded_id(self) -> int | str | None: """Ensure that the ID is url-encoded so that it can be safely used in a URL path""" obj_id = self.get_id() @@ -280,7 +278,7 @@ class RESTObjectList: """ def __init__( - self, manager: "RESTManager[Any]", obj_cls: Type[RESTObject], _list: GitlabList + self, manager: RESTManager[Any], obj_cls: type[RESTObject], _list: GitlabList ) -> None: """Creates an objects list from a GitlabList. @@ -296,7 +294,7 @@ def __init__( self._obj_cls = obj_cls self._list = _list - def __iter__(self) -> "RESTObjectList": + def __iter__(self) -> RESTObjectList: return self def __len__(self) -> int: @@ -315,7 +313,7 @@ def current_page(self) -> int: return self._list.current_page @property - def prev_page(self) -> Optional[int]: + def prev_page(self) -> int | None: """The previous page number. If None, the current page is the first. @@ -323,7 +321,7 @@ def prev_page(self) -> Optional[int]: return self._list.prev_page @property - def next_page(self) -> Optional[int]: + def next_page(self) -> int | None: """The next page number. If None, the current page is the last. @@ -331,17 +329,17 @@ def next_page(self) -> Optional[int]: return self._list.next_page @property - def per_page(self) -> Optional[int]: + def per_page(self) -> int | None: """The number of items per page.""" return self._list.per_page @property - def total_pages(self) -> Optional[int]: + def total_pages(self) -> int | None: """The total number of pages.""" return self._list.total_pages @property - def total(self) -> Optional[int]: + def total(self) -> int | None: """The total number of items.""" return self._list.total @@ -362,15 +360,15 @@ class RESTManager(Generic[TObjCls]): _update_attrs: g_types.RequiredOptional = g_types.RequiredOptional() _path: ClassVar[str] _obj_cls: type[TObjCls] - _from_parent_attrs: Dict[str, Any] = {} - _types: Dict[str, Type[g_types.GitlabAttribute]] = {} + _from_parent_attrs: dict[str, Any] = {} + _types: dict[str, type[g_types.GitlabAttribute]] = {} _computed_path: str - _parent: Optional[RESTObject] - _parent_attrs: Dict[str, Any] + _parent: RESTObject | None + _parent_attrs: dict[str, Any] gitlab: Gitlab - def __init__(self, gl: Gitlab, parent: Optional[RESTObject] = None) -> None: + def __init__(self, gl: Gitlab, parent: RESTObject | None = None) -> None: """REST manager constructor. Args: @@ -382,17 +380,17 @@ def __init__(self, gl: Gitlab, parent: Optional[RESTObject] = None) -> None: self._computed_path = self._compute_path() @property - def parent_attrs(self) -> Optional[Dict[str, Any]]: + def parent_attrs(self) -> dict[str, Any] | None: return self._parent_attrs - def _compute_path(self, path: Optional[str] = None) -> str: + def _compute_path(self, path: str | None = None) -> str: self._parent_attrs = {} if path is None: path = self._path if self._parent is None or not self._from_parent_attrs: return path - data: Dict[str, Optional[gitlab.utils.EncodedId]] = {} + data: dict[str, gitlab.utils.EncodedId | None] = {} for self_attr, parent_attr in self._from_parent_attrs.items(): if not hasattr(self._parent, parent_attr): data[self_attr] = None diff --git a/gitlab/cli.py b/gitlab/cli.py index fa139a7d5..0e1d56cb5 100644 --- a/gitlab/cli.py +++ b/gitlab/cli.py @@ -1,3 +1,5 @@ +from __future__ import annotations + import argparse import dataclasses import functools @@ -10,14 +12,9 @@ Any, Callable, cast, - Dict, NoReturn, - Optional, - Tuple, - Type, TYPE_CHECKING, TypeVar, - Union, ) from requests.structures import CaseInsensitiveDict @@ -33,11 +30,11 @@ @dataclasses.dataclass class CustomAction: - required: Tuple[str, ...] - optional: Tuple[str, ...] + required: tuple[str, ...] + optional: tuple[str, ...] in_object: bool requires_id: bool # if the `_id_attr` value should be a required argument - help: Optional[str] # help text for the custom action + help: str | None # help text for the custom action # custom_actions = { @@ -45,7 +42,7 @@ class CustomAction: # action: CustomAction, # }, # } -custom_actions: Dict[str, Dict[str, CustomAction]] = {} +custom_actions: dict[str, dict[str, CustomAction]] = {} # For an explanation of how these type-hints work see: @@ -57,12 +54,12 @@ class CustomAction: def register_custom_action( *, - cls_names: Union[str, Tuple[str, ...]], - required: Tuple[str, ...] = (), - optional: Tuple[str, ...] = (), - custom_action: Optional[str] = None, + cls_names: str | tuple[str, ...], + required: tuple[str, ...] = (), + optional: tuple[str, ...] = (), + custom_action: str | None = None, requires_id: bool = True, # if the `_id_attr` value should be a required argument - help: Optional[str] = None, # help text for the action + help: str | None = None, # help text for the action ) -> Callable[[__F], __F]: def wrap(f: __F) -> __F: @functools.wraps(f) @@ -98,7 +95,7 @@ def wrapped_f(*args: Any, **kwargs: Any) -> Any: return wrap -def die(msg: str, e: Optional[Exception] = None) -> NoReturn: +def die(msg: str, e: Exception | None = None) -> NoReturn: if e: msg = f"{msg} ({e})" sys.stderr.write(f"{msg}\n") @@ -107,7 +104,7 @@ def die(msg: str, e: Optional[Exception] = None) -> NoReturn: def gitlab_resource_to_cls( gitlab_resource: str, namespace: ModuleType -) -> Type[RESTObject]: +) -> type[RESTObject]: classes = CaseInsensitiveDict(namespace.__dict__) lowercase_class = gitlab_resource.replace("-", "") class_type = classes[lowercase_class] diff --git a/gitlab/client.py b/gitlab/client.py index 87b324c34..b41f7a885 100644 --- a/gitlab/client.py +++ b/gitlab/client.py @@ -8,13 +8,7 @@ Any, BinaryIO, cast, - Dict, - List, - Optional, - Tuple, - Type, TYPE_CHECKING, - Union, ) from urllib import parse @@ -83,26 +77,26 @@ class Gitlab: def __init__( self, - url: Optional[str] = None, - private_token: Optional[str] = None, - oauth_token: Optional[str] = None, - job_token: Optional[str] = None, - ssl_verify: Union[bool, str] = True, - http_username: Optional[str] = None, - http_password: Optional[str] = None, - timeout: Optional[float] = None, + url: str | None = None, + private_token: str | None = None, + oauth_token: str | None = None, + job_token: str | None = None, + ssl_verify: bool | str = True, + http_username: str | None = None, + http_password: str | None = None, + timeout: float | None = None, api_version: str = "4", - per_page: Optional[int] = None, - pagination: Optional[str] = None, - order_by: Optional[str] = None, + per_page: int | None = None, + pagination: str | None = None, + order_by: str | None = None, user_agent: str = gitlab.const.USER_AGENT, retry_transient_errors: bool = False, keep_base_url: bool = False, **kwargs: Any, ) -> None: self._api_version = str(api_version) - self._server_version: Optional[str] = None - self._server_revision: Optional[str] = None + self._server_version: str | None = None + self._server_revision: str | None = None self._base_url = utils.get_base_url(url) self._url = f"{self._base_url}/api/v{api_version}" #: Timeout to use for requests to gitlab server @@ -123,7 +117,7 @@ def __init__( self._set_auth_info() #: Create a session object for requests - _backend: Type[_backends.DefaultBackend] = kwargs.pop( + _backend: type[_backends.DefaultBackend] = kwargs.pop( "backend", _backends.DefaultBackend ) self._backend = _backend(**kwargs) @@ -141,7 +135,7 @@ def __init__( from gitlab.v4 import objects self._objects = objects - self.user: Optional[objects.CurrentUser] = None + self.user: objects.CurrentUser | None = None self.broadcastmessages = objects.BroadcastMessageManager(self) """See :class:`~gitlab.v4.objects.BroadcastMessageManager`""" @@ -224,18 +218,18 @@ def __init__( self.statistics = objects.ApplicationStatisticsManager(self) """See :class:`~gitlab.v4.objects.ApplicationStatisticsManager`""" - def __enter__(self) -> "Gitlab": + def __enter__(self) -> Gitlab: return self def __exit__(self, *args: Any) -> None: self.session.close() - def __getstate__(self) -> Dict[str, Any]: + def __getstate__(self) -> dict[str, Any]: state = self.__dict__.copy() state.pop("_objects") return state - def __setstate__(self, state: Dict[str, Any]) -> None: + def __setstate__(self, state: dict[str, Any]) -> None: self.__dict__.update(state) # We only support v4 API at this time if self._api_version not in ("4",): @@ -266,10 +260,10 @@ def api_version(self) -> str: @classmethod def from_config( cls, - gitlab_id: Optional[str] = None, - config_files: Optional[List[str]] = None, + gitlab_id: str | None = None, + config_files: list[str] | None = None, **kwargs: Any, - ) -> "Gitlab": + ) -> Gitlab: """Create a Gitlab connection from configuration files. Args: @@ -310,10 +304,10 @@ def from_config( @classmethod def merge_config( cls, - options: Dict[str, Any], - gitlab_id: Optional[str] = None, - config_files: Optional[List[str]] = None, - ) -> "Gitlab": + options: dict[str, Any], + gitlab_id: str | None = None, + config_files: list[str] | None = None, + ) -> Gitlab: """Create a Gitlab connection by merging configuration with the following precedence: @@ -364,8 +358,8 @@ def merge_config( @staticmethod def _merge_auth( - options: Dict[str, Any], config: gitlab.config.GitlabConfigParser - ) -> Tuple[Optional[str], Optional[str], Optional[str]]: + options: dict[str, Any], config: gitlab.config.GitlabConfigParser + ) -> tuple[str | None, str | None, str | None]: """ Return a tuple where at most one of 3 token types ever has a value. Since multiple types of tokens may be present in the environment, @@ -402,7 +396,7 @@ def auth(self) -> None: if hasattr(self.user, "web_url") and hasattr(self.user, "username"): self._check_url(self.user.web_url, path=self.user.username) - def version(self) -> Tuple[str, str]: + def version(self) -> tuple[str, str]: """Returns the version and revision of the gitlab server. Note that self.version and self.revision will be set on the gitlab @@ -429,7 +423,7 @@ def version(self) -> Tuple[str, str]: @gitlab.exceptions.on_http_error(gitlab.exceptions.GitlabMarkdownError) def markdown( - self, text: str, gfm: bool = False, project: Optional[str] = None, **kwargs: Any + self, text: str, gfm: bool = False, project: str | None = None, **kwargs: Any ) -> str: """Render an arbitrary Markdown document. @@ -456,7 +450,7 @@ def markdown( return data["html"] @gitlab.exceptions.on_http_error(gitlab.exceptions.GitlabLicenseError) - def get_license(self, **kwargs: Any) -> Dict[str, Union[str, Dict[str, str]]]: + def get_license(self, **kwargs: Any) -> dict[str, str | dict[str, str]]: """Retrieve information about the current license. Args: @@ -475,7 +469,7 @@ def get_license(self, **kwargs: Any) -> Dict[str, Union[str, Dict[str, str]]]: return {} @gitlab.exceptions.on_http_error(gitlab.exceptions.GitlabLicenseError) - def set_license(self, license: str, **kwargs: Any) -> Dict[str, Any]: + def set_license(self, license: str, **kwargs: Any) -> dict[str, Any]: """Add a new license. Args: @@ -516,7 +510,7 @@ def _set_auth_info(self) -> None: "authentication should be defined" ) - self._auth: Optional[requests.auth.AuthBase] = None + self._auth: requests.auth.AuthBase | None = None if self.private_token: self._auth = _backends.PrivateTokenAuth(self.private_token) @@ -564,7 +558,7 @@ def print_as_log(*args: Any) -> None: logger.handlers.clear() logger.addHandler(handler) - def _get_session_opts(self) -> Dict[str, Any]: + def _get_session_opts(self) -> dict[str, Any]: return { "headers": self.headers.copy(), "auth": self._auth, @@ -585,7 +579,7 @@ def _build_url(self, path: str) -> str: return path return f"{self._url}{path}" - def _check_url(self, url: Optional[str], *, path: str = "api") -> Optional[str]: + def _check_url(self, url: str | None, *, path: str = "api") -> str | None: """ Checks if ``url`` starts with a different base URL from the user-provided base URL and warns the user before returning it. If ``keep_base_url`` is set to @@ -645,16 +639,16 @@ def http_request( self, verb: str, path: str, - query_data: Optional[Dict[str, Any]] = None, - post_data: Optional[Union[Dict[str, Any], bytes, BinaryIO]] = None, + query_data: dict[str, Any] | None = None, + post_data: dict[str, Any] | bytes | BinaryIO | None = None, raw: bool = False, streamed: bool = False, - files: Optional[Dict[str, Any]] = None, - timeout: Optional[float] = None, + files: dict[str, Any] | None = None, + timeout: float | None = None, obey_rate_limit: bool = True, - retry_transient_errors: Optional[bool] = None, + retry_transient_errors: bool | None = None, max_retries: int = 10, - extra_headers: Optional[Dict[str, Any]] = None, + extra_headers: dict[str, Any] | None = None, **kwargs: Any, ) -> requests.Response: """Make an HTTP request to the Gitlab server. @@ -785,11 +779,11 @@ def http_request( def http_get( self, path: str, - query_data: Optional[Dict[str, Any]] = None, + query_data: dict[str, Any] | None = None, streamed: bool = False, raw: bool = False, **kwargs: Any, - ) -> Union[Dict[str, Any], requests.Response]: + ) -> dict[str, Any] | requests.Response: """Make a GET request to the Gitlab server. Args: @@ -829,8 +823,8 @@ def http_get( return result def http_head( - self, path: str, query_data: Optional[Dict[str, Any]] = None, **kwargs: Any - ) -> "requests.structures.CaseInsensitiveDict[Any]": + self, path: str, query_data: dict[str, Any] | None = None, **kwargs: Any + ) -> requests.structures.CaseInsensitiveDict[Any]: """Make a HEAD request to the Gitlab server. Args: @@ -852,12 +846,12 @@ def http_head( def http_list( self, path: str, - query_data: Optional[Dict[str, Any]] = None, + query_data: dict[str, Any] | None = None, *, - iterator: Optional[bool] = None, - message_details: Optional[utils.WarnMessageData] = None, + iterator: bool | None = None, + message_details: utils.WarnMessageData | None = None, **kwargs: Any, - ) -> Union["GitlabList", List[Dict[str, Any]]]: + ) -> GitlabList | list[dict[str, Any]]: """Make a GET request to the Gitlab server for list-oriented queries. Args: @@ -968,12 +962,12 @@ def should_emit_warning() -> bool: def http_post( self, path: str, - query_data: Optional[Dict[str, Any]] = None, - post_data: Optional[Dict[str, Any]] = None, + query_data: dict[str, Any] | None = None, + post_data: dict[str, Any] | None = None, raw: bool = False, - files: Optional[Dict[str, Any]] = None, + files: dict[str, Any] | None = None, **kwargs: Any, - ) -> Union[Dict[str, Any], requests.Response]: + ) -> dict[str, Any] | requests.Response: """Make a POST request to the Gitlab server. Args: @@ -1023,12 +1017,12 @@ def http_post( def http_put( self, path: str, - query_data: Optional[Dict[str, Any]] = None, - post_data: Optional[Union[Dict[str, Any], bytes, BinaryIO]] = None, + query_data: dict[str, Any] | None = None, + post_data: dict[str, Any] | bytes | BinaryIO | None = None, raw: bool = False, - files: Optional[Dict[str, Any]] = None, + files: dict[str, Any] | None = None, **kwargs: Any, - ) -> Union[Dict[str, Any], requests.Response]: + ) -> dict[str, Any] | requests.Response: """Make a PUT request to the Gitlab server. Args: @@ -1076,11 +1070,11 @@ def http_patch( self, path: str, *, - query_data: Optional[Dict[str, Any]] = None, - post_data: Optional[Union[Dict[str, Any], bytes]] = None, + query_data: dict[str, Any] | None = None, + post_data: dict[str, Any] | bytes | None = None, raw: bool = False, **kwargs: Any, - ) -> Union[Dict[str, Any], requests.Response]: + ) -> dict[str, Any] | requests.Response: """Make a PATCH request to the Gitlab server. Args: @@ -1141,7 +1135,7 @@ def http_delete(self, path: str, **kwargs: Any) -> requests.Response: @gitlab.exceptions.on_http_error(gitlab.exceptions.GitlabSearchError) def search( self, scope: str, search: str, **kwargs: Any - ) -> Union["GitlabList", List[Dict[str, Any]]]: + ) -> GitlabList | list[dict[str, Any]]: """Search GitLab resources matching the provided string.' Args: @@ -1171,7 +1165,7 @@ def __init__( self, gl: Gitlab, url: str, - query_data: Dict[str, Any], + query_data: dict[str, Any], get_next: bool = True, **kwargs: Any, ) -> None: @@ -1187,7 +1181,7 @@ def __init__( self._kwargs.pop("query_parameters", None) def _query( - self, url: str, query_data: Optional[Dict[str, Any]] = None, **kwargs: Any + self, url: str, query_data: dict[str, Any] | None = None, **kwargs: Any ) -> None: query_data = query_data or {} result = self._gl.http_request("get", url, query_data=query_data, **kwargs) @@ -1197,15 +1191,15 @@ def _query( next_url = None self._next_url = self._gl._check_url(next_url) - self._current_page: Optional[str] = result.headers.get("X-Page") - self._prev_page: Optional[str] = result.headers.get("X-Prev-Page") - self._next_page: Optional[str] = result.headers.get("X-Next-Page") - self._per_page: Optional[str] = result.headers.get("X-Per-Page") - self._total_pages: Optional[str] = result.headers.get("X-Total-Pages") - self._total: Optional[str] = result.headers.get("X-Total") + self._current_page: str | None = result.headers.get("X-Page") + self._prev_page: str | None = result.headers.get("X-Prev-Page") + self._next_page: str | None = result.headers.get("X-Next-Page") + self._per_page: str | None = result.headers.get("X-Per-Page") + self._total_pages: str | None = result.headers.get("X-Total-Pages") + self._total: str | None = result.headers.get("X-Total") try: - self._data: List[Dict[str, Any]] = result.json() + self._data: list[dict[str, Any]] = result.json() except Exception as e: raise gitlab.exceptions.GitlabParsingError( error_message="Failed to parse the server message" @@ -1221,7 +1215,7 @@ def current_page(self) -> int: return int(self._current_page) @property - def prev_page(self) -> Optional[int]: + def prev_page(self) -> int | None: """The previous page number. If None, the current page is the first. @@ -1229,7 +1223,7 @@ def prev_page(self) -> Optional[int]: return int(self._prev_page) if self._prev_page else None @property - def next_page(self) -> Optional[int]: + def next_page(self) -> int | None: """The next page number. If None, the current page is the last. @@ -1237,7 +1231,7 @@ def next_page(self) -> Optional[int]: return int(self._next_page) if self._next_page else None @property - def per_page(self) -> Optional[int]: + def per_page(self) -> int | None: """The number of items per page.""" return int(self._per_page) if self._per_page is not None else None @@ -1245,20 +1239,20 @@ def per_page(self) -> Optional[int]: # the headers 'x-total-pages' and 'x-total'. In those cases we return None. # https://docs.gitlab.com/ee/user/gitlab_com/index.html#pagination-response-headers @property - def total_pages(self) -> Optional[int]: + def total_pages(self) -> int | None: """The total number of pages.""" if self._total_pages is not None: return int(self._total_pages) return None @property - def total(self) -> Optional[int]: + def total(self) -> int | None: """The total number of items.""" if self._total is not None: return int(self._total) return None - def __iter__(self) -> "GitlabList": + def __iter__(self) -> GitlabList: return self def __len__(self) -> int: @@ -1266,10 +1260,10 @@ def __len__(self) -> int: return 0 return int(self._total) - def __next__(self) -> Dict[str, Any]: + def __next__(self) -> dict[str, Any]: return self.next() - def next(self) -> Dict[str, Any]: + def next(self) -> dict[str, Any]: try: item = self._data[self._current] self._current += 1 @@ -1287,11 +1281,11 @@ def next(self) -> Dict[str, Any]: class _BaseGraphQL: def __init__( self, - url: Optional[str] = None, + url: str | None = None, *, - token: Optional[str] = None, - ssl_verify: Union[bool, str] = True, - timeout: Optional[float] = None, + token: str | None = None, + ssl_verify: bool | str = True, + timeout: float | None = None, user_agent: str = gitlab.const.USER_AGENT, fetch_schema_from_transport: bool = False, max_retries: int = 10, @@ -1316,7 +1310,7 @@ def __init__( self._client_opts = self._get_client_opts() self._fetch_schema_from_transport = fetch_schema_from_transport - def _get_client_opts(self) -> Dict[str, Any]: + def _get_client_opts(self) -> dict[str, Any]: headers = {"User-Agent": self._user_agent} if self._token: @@ -1332,12 +1326,12 @@ def _get_client_opts(self) -> Dict[str, Any]: class GraphQL(_BaseGraphQL): def __init__( self, - url: Optional[str] = None, + url: str | None = None, *, - token: Optional[str] = None, - ssl_verify: Union[bool, str] = True, - client: Optional[httpx.Client] = None, - timeout: Optional[float] = None, + token: str | None = None, + ssl_verify: bool | str = True, + client: httpx.Client | None = None, + timeout: float | None = None, user_agent: str = gitlab.const.USER_AGENT, fetch_schema_from_transport: bool = False, max_retries: int = 10, @@ -1364,15 +1358,13 @@ def __init__( ) self._gql = gql.gql - def __enter__(self) -> "GraphQL": + def __enter__(self) -> GraphQL: return self def __exit__(self, *args: Any) -> None: self._http_client.close() - def execute( - self, request: Union[str, graphql.Source], *args: Any, **kwargs: Any - ) -> Any: + def execute(self, request: str | graphql.Source, *args: Any, **kwargs: Any) -> Any: parsed_document = self._gql(request) retry = utils.Retry( max_retries=self._max_retries, @@ -1406,12 +1398,12 @@ def execute( class AsyncGraphQL(_BaseGraphQL): def __init__( self, - url: Optional[str] = None, + url: str | None = None, *, - token: Optional[str] = None, - ssl_verify: Union[bool, str] = True, - client: Optional[httpx.AsyncClient] = None, - timeout: Optional[float] = None, + token: str | None = None, + ssl_verify: bool | str = True, + client: httpx.AsyncClient | None = None, + timeout: float | None = None, user_agent: str = gitlab.const.USER_AGENT, fetch_schema_from_transport: bool = False, max_retries: int = 10, @@ -1438,14 +1430,14 @@ def __init__( ) self._gql = gql.gql - async def __aenter__(self) -> "AsyncGraphQL": + async def __aenter__(self) -> AsyncGraphQL: return self async def __aexit__(self, *args: Any) -> None: await self._http_client.aclose() async def execute( - self, request: Union[str, graphql.Source], *args: Any, **kwargs: Any + self, request: str | graphql.Source, *args: Any, **kwargs: Any ) -> Any: parsed_document = self._gql(request) retry = utils.Retry( diff --git a/gitlab/config.py b/gitlab/config.py index 0f4b2cd6e..a0f207af9 100644 --- a/gitlab/config.py +++ b/gitlab/config.py @@ -1,14 +1,15 @@ +from __future__ import annotations + import configparser import os import shlex import subprocess from os.path import expanduser, expandvars from pathlib import Path -from typing import List, Optional, Union from gitlab.const import USER_AGENT -_DEFAULT_FILES: List[str] = [ +_DEFAULT_FILES: list[str] = [ "/etc/python-gitlab.cfg", str(Path.home() / ".python-gitlab.cfg"), ] @@ -20,14 +21,14 @@ _CONFIG_PARSER_ERRORS = (configparser.NoOptionError, configparser.NoSectionError) -def _resolve_file(filepath: Union[Path, str]) -> str: +def _resolve_file(filepath: Path | str) -> str: resolved = Path(filepath).resolve(strict=True) return str(resolved) def _get_config_files( - config_files: Optional[List[str]] = None, -) -> Union[str, List[str]]: + config_files: list[str] | None = None, +) -> str | list[str]: """ Return resolved path(s) to config files if they exist, with precedence: 1. Files passed in config_files @@ -90,23 +91,23 @@ class GitlabConfigHelperError(ConfigError): class GitlabConfigParser: def __init__( - self, gitlab_id: Optional[str] = None, config_files: Optional[List[str]] = None + self, gitlab_id: str | None = None, config_files: list[str] | None = None ) -> None: self.gitlab_id = gitlab_id - self.http_username: Optional[str] = None - self.http_password: Optional[str] = None - self.job_token: Optional[str] = None - self.oauth_token: Optional[str] = None - self.private_token: Optional[str] = None + self.http_username: str | None = None + self.http_password: str | None = None + self.job_token: str | None = None + self.oauth_token: str | None = None + self.private_token: str | None = None self.api_version: str = "4" - self.order_by: Optional[str] = None - self.pagination: Optional[str] = None - self.per_page: Optional[int] = None + self.order_by: str | None = None + self.pagination: str | None = None + self.per_page: int | None = None self.retry_transient_errors: bool = False - self.ssl_verify: Union[bool, str] = True + self.ssl_verify: bool | str = True self.timeout: int = 60 - self.url: Optional[str] = None + self.url: str | None = None self.user_agent: str = USER_AGENT self.keep_base_url: bool = False diff --git a/gitlab/exceptions.py b/gitlab/exceptions.py index 35f7dc11c..7aa42152c 100644 --- a/gitlab/exceptions.py +++ b/gitlab/exceptions.py @@ -1,13 +1,15 @@ +from __future__ import annotations + import functools -from typing import Any, Callable, cast, Optional, Type, TYPE_CHECKING, TypeVar, Union +from typing import Any, Callable, cast, TYPE_CHECKING, TypeVar class GitlabError(Exception): def __init__( self, - error_message: Union[str, bytes] = "", - response_code: Optional[int] = None, - response_body: Optional[bytes] = None, + error_message: str | bytes = "", + response_code: int | None = None, + response_body: bytes | None = None, ) -> None: Exception.__init__(self, error_message) # Http status code @@ -327,7 +329,7 @@ class GitlabHookTestError(GitlabOperationError): __F = TypeVar("__F", bound=Callable[..., Any]) -def on_http_error(error: Type[Exception]) -> Callable[[__F], __F]: +def on_http_error(error: type[Exception]) -> Callable[[__F], __F]: """Manage GitlabHttpError exceptions. This decorator function can be used to catch GitlabHttpError exceptions diff --git a/gitlab/mixins.py b/gitlab/mixins.py index df4caa40b..e761fbe04 100644 --- a/gitlab/mixins.py +++ b/gitlab/mixins.py @@ -1,17 +1,14 @@ +from __future__ import annotations + import enum +from collections.abc import Iterator from types import ModuleType from typing import ( Any, Callable, - Dict, - Iterator, - List, Literal, - Optional, overload, - Tuple, TYPE_CHECKING, - Union, ) import requests @@ -55,8 +52,8 @@ class HeadMixin(base.RESTManager[base.TObjCls]): @exc.on_http_error(exc.GitlabHeadError) def head( - self, id: Optional[Union[str, int]] = None, **kwargs: Any - ) -> "requests.structures.CaseInsensitiveDict[Any]": + self, id: str | int | None = None, **kwargs: Any + ) -> requests.structures.CaseInsensitiveDict[Any]: """Retrieve headers from an endpoint. Args: @@ -78,12 +75,10 @@ def head( class GetMixin(HeadMixin[base.TObjCls]): - _optional_get_attrs: Tuple[str, ...] = () + _optional_get_attrs: tuple[str, ...] = () @exc.on_http_error(exc.GitlabGetError) - def get( - self, id: Union[str, int], lazy: bool = False, **kwargs: Any - ) -> base.TObjCls: + def get(self, id: str | int, lazy: bool = False, **kwargs: Any) -> base.TObjCls: """Retrieve a single object. Args: @@ -114,7 +109,7 @@ def get( class GetWithoutIdMixin(HeadMixin[base.TObjCls]): - _optional_get_attrs: Tuple[str, ...] = () + _optional_get_attrs: tuple[str, ...] = () @exc.on_http_error(exc.GitlabGetError) def get(self, **kwargs: Any) -> base.TObjCls: @@ -137,11 +132,11 @@ def get(self, **kwargs: Any) -> base.TObjCls: class RefreshMixin(_RestObjectBase): - _id_attr: Optional[str] - _attrs: Dict[str, Any] + _id_attr: str | None + _attrs: dict[str, Any] _module: ModuleType - _parent_attrs: Dict[str, Any] - _updated_attrs: Dict[str, Any] + _parent_attrs: dict[str, Any] + _updated_attrs: dict[str, Any] manager: base.RESTManager[Any] @exc.on_http_error(exc.GitlabGetError) @@ -170,10 +165,10 @@ def refresh(self, **kwargs: Any) -> None: class ListMixin(HeadMixin[base.TObjCls]): - _list_filters: Tuple[str, ...] = () + _list_filters: tuple[str, ...] = () @exc.on_http_error(exc.GitlabListError) - def list(self, **kwargs: Any) -> Union[base.RESTObjectList, List[base.TObjCls]]: + def list(self, **kwargs: Any) -> base.RESTObjectList | list[base.TObjCls]: """Retrieve a list of objects. Args: @@ -223,9 +218,7 @@ class RetrieveMixin(ListMixin[base.TObjCls], GetMixin[base.TObjCls]): ... class CreateMixin(base.RESTManager[base.TObjCls]): @exc.on_http_error(exc.GitlabCreateError) - def create( - self, data: Optional[Dict[str, Any]] = None, **kwargs: Any - ) -> base.TObjCls: + def create(self, data: dict[str, Any] | None = None, **kwargs: Any) -> base.TObjCls: """Create a new object. Args: @@ -270,7 +263,7 @@ class UpdateMixin(base.RESTManager[base.TObjCls]): def _get_update_method( self, - ) -> Callable[..., Union[Dict[str, Any], "requests.Response"]]: + ) -> Callable[..., dict[str, Any] | requests.Response]: """Return the HTTP method to use. Returns: @@ -288,10 +281,10 @@ def _get_update_method( @exc.on_http_error(exc.GitlabUpdateError) def update( self, - id: Optional[Union[str, int]] = None, - new_data: Optional[Dict[str, Any]] = None, + id: str | int | None = None, + new_data: dict[str, Any] | None = None, **kwargs: Any, - ) -> Dict[str, Any]: + ) -> dict[str, Any]: """Update an object on the server. Args: @@ -355,7 +348,7 @@ def set(self, key: str, value: str, **kwargs: Any) -> base.TObjCls: class DeleteMixin(base.RESTManager[base.TObjCls]): @exc.on_http_error(exc.GitlabDeleteError) - def delete(self, id: Optional[Union[str, int]] = None, **kwargs: Any) -> None: + def delete(self, id: str | int | None = None, **kwargs: Any) -> None: """Delete an object on the server. Args: @@ -394,14 +387,14 @@ class NoUpdateMixin( class SaveMixin(_RestObjectBase): """Mixin for RESTObject's that can be updated.""" - _id_attr: Optional[str] - _attrs: Dict[str, Any] + _id_attr: str | None + _attrs: dict[str, Any] _module: ModuleType - _parent_attrs: Dict[str, Any] - _updated_attrs: Dict[str, Any] + _parent_attrs: dict[str, Any] + _updated_attrs: dict[str, Any] manager: base.RESTManager[Any] - def _get_updated_data(self) -> Dict[str, Any]: + def _get_updated_data(self) -> dict[str, Any]: updated_data = {} for attr in self.manager._update_attrs.required: # Get everything required, no matter if it's been updated @@ -411,7 +404,7 @@ def _get_updated_data(self) -> Dict[str, Any]: return updated_data - def save(self, **kwargs: Any) -> Optional[Dict[str, Any]]: + def save(self, **kwargs: Any) -> dict[str, Any] | None: """Save the changes made to the object to the server. The object is updated to match what the server returns. @@ -443,11 +436,11 @@ def save(self, **kwargs: Any) -> Optional[Dict[str, Any]]: class ObjectDeleteMixin(_RestObjectBase): """Mixin for RESTObject's that can be deleted.""" - _id_attr: Optional[str] - _attrs: Dict[str, Any] + _id_attr: str | None + _attrs: dict[str, Any] _module: ModuleType - _parent_attrs: Dict[str, Any] - _updated_attrs: Dict[str, Any] + _parent_attrs: dict[str, Any] + _updated_attrs: dict[str, Any] manager: base.RESTManager[Any] def delete(self, **kwargs: Any) -> None: @@ -467,16 +460,16 @@ def delete(self, **kwargs: Any) -> None: class UserAgentDetailMixin(_RestObjectBase): - _id_attr: Optional[str] - _attrs: Dict[str, Any] + _id_attr: str | None + _attrs: dict[str, Any] _module: ModuleType - _parent_attrs: Dict[str, Any] - _updated_attrs: Dict[str, Any] + _parent_attrs: dict[str, Any] + _updated_attrs: dict[str, Any] manager: base.RESTManager[Any] @cli.register_custom_action(cls_names=("Snippet", "ProjectSnippet", "ProjectIssue")) @exc.on_http_error(exc.GitlabGetError) - def user_agent_detail(self, **kwargs: Any) -> Dict[str, Any]: + def user_agent_detail(self, **kwargs: Any) -> dict[str, Any]: """Get the user agent detail. Args: @@ -494,11 +487,11 @@ def user_agent_detail(self, **kwargs: Any) -> Dict[str, Any]: class AccessRequestMixin(_RestObjectBase): - _id_attr: Optional[str] - _attrs: Dict[str, Any] + _id_attr: str | None + _attrs: dict[str, Any] _module: ModuleType - _parent_attrs: Dict[str, Any] - _updated_attrs: Dict[str, Any] + _parent_attrs: dict[str, Any] + _updated_attrs: dict[str, Any] manager: base.RESTManager[Any] @cli.register_custom_action( @@ -529,11 +522,11 @@ def approve( class DownloadMixin(_RestObjectBase): - _id_attr: Optional[str] - _attrs: Dict[str, Any] + _id_attr: str | None + _attrs: dict[str, Any] _module: ModuleType - _parent_attrs: Dict[str, Any] - _updated_attrs: Dict[str, Any] + _parent_attrs: dict[str, Any] + _updated_attrs: dict[str, Any] manager: base.RESTManager[Any] @overload @@ -562,7 +555,7 @@ def download( def download( self, streamed: Literal[True] = True, - action: Optional[Callable[[bytes], Any]] = None, + action: Callable[[bytes], Any] | None = None, chunk_size: int = 1024, *, iterator: Literal[False] = False, @@ -574,12 +567,12 @@ def download( def download( self, streamed: bool = False, - action: Optional[Callable[[bytes], Any]] = None, + action: Callable[[bytes], Any] | None = None, chunk_size: int = 1024, *, iterator: bool = False, **kwargs: Any, - ) -> Optional[Union[bytes, Iterator[Any]]]: + ) -> bytes | Iterator[Any] | None: """Download the archive of a resource export. Args: @@ -622,11 +615,8 @@ class RotateMixin(base.RESTManager[base.TObjCls]): ) @exc.on_http_error(exc.GitlabRotateError) def rotate( - self, - id: Union[str, int], - expires_at: Optional[str] = None, - **kwargs: Any, - ) -> Dict[str, Any]: + self, id: str | int, expires_at: str | None = None, **kwargs: Any + ) -> dict[str, Any]: """Rotate an access token. Args: @@ -638,7 +628,7 @@ def rotate( GitlabRotateError: If the server cannot perform the request """ path = f"{self.path}/{utils.EncodedId(id)}/rotate" - data: Dict[str, Any] = {} + data: dict[str, Any] = {} if expires_at is not None: data = {"expires_at": expires_at} @@ -649,11 +639,11 @@ def rotate( class ObjectRotateMixin(_RestObjectBase): - _id_attr: Optional[str] - _attrs: Dict[str, Any] + _id_attr: str | None + _attrs: dict[str, Any] _module: ModuleType - _parent_attrs: Dict[str, Any] - _updated_attrs: Dict[str, Any] + _parent_attrs: dict[str, Any] + _updated_attrs: dict[str, Any] manager: base.RESTManager[Any] @cli.register_custom_action( @@ -661,7 +651,7 @@ class ObjectRotateMixin(_RestObjectBase): optional=("expires_at",), ) @exc.on_http_error(exc.GitlabRotateError) - def rotate(self, **kwargs: Any) -> Dict[str, Any]: + def rotate(self, **kwargs: Any) -> dict[str, Any]: """Rotate the current access token object. Args: @@ -680,11 +670,11 @@ def rotate(self, **kwargs: Any) -> Dict[str, Any]: class SubscribableMixin(_RestObjectBase): - _id_attr: Optional[str] - _attrs: Dict[str, Any] + _id_attr: str | None + _attrs: dict[str, Any] _module: ModuleType - _parent_attrs: Dict[str, Any] - _updated_attrs: Dict[str, Any] + _parent_attrs: dict[str, Any] + _updated_attrs: dict[str, Any] manager: base.RESTManager[Any] @cli.register_custom_action( @@ -729,11 +719,11 @@ def unsubscribe(self, **kwargs: Any) -> None: class TodoMixin(_RestObjectBase): - _id_attr: Optional[str] - _attrs: Dict[str, Any] + _id_attr: str | None + _attrs: dict[str, Any] _module: ModuleType - _parent_attrs: Dict[str, Any] - _updated_attrs: Dict[str, Any] + _parent_attrs: dict[str, Any] + _updated_attrs: dict[str, Any] manager: base.RESTManager[Any] @cli.register_custom_action(cls_names=("ProjectIssue", "ProjectMergeRequest")) @@ -753,16 +743,16 @@ def todo(self, **kwargs: Any) -> None: class TimeTrackingMixin(_RestObjectBase): - _id_attr: Optional[str] - _attrs: Dict[str, Any] + _id_attr: str | None + _attrs: dict[str, Any] _module: ModuleType - _parent_attrs: Dict[str, Any] - _updated_attrs: Dict[str, Any] + _parent_attrs: dict[str, Any] + _updated_attrs: dict[str, Any] manager: base.RESTManager[Any] @cli.register_custom_action(cls_names=("ProjectIssue", "ProjectMergeRequest")) @exc.on_http_error(exc.GitlabTimeTrackingError) - def time_stats(self, **kwargs: Any) -> Dict[str, Any]: + def time_stats(self, **kwargs: Any) -> dict[str, Any]: """Get time stats for the object. Args: @@ -790,7 +780,7 @@ def time_stats(self, **kwargs: Any) -> Dict[str, Any]: cls_names=("ProjectIssue", "ProjectMergeRequest"), required=("duration",) ) @exc.on_http_error(exc.GitlabTimeTrackingError) - def time_estimate(self, duration: str, **kwargs: Any) -> Dict[str, Any]: + def time_estimate(self, duration: str, **kwargs: Any) -> dict[str, Any]: """Set an estimated time of work for the object. Args: @@ -810,7 +800,7 @@ def time_estimate(self, duration: str, **kwargs: Any) -> Dict[str, Any]: @cli.register_custom_action(cls_names=("ProjectIssue", "ProjectMergeRequest")) @exc.on_http_error(exc.GitlabTimeTrackingError) - def reset_time_estimate(self, **kwargs: Any) -> Dict[str, Any]: + def reset_time_estimate(self, **kwargs: Any) -> dict[str, Any]: """Resets estimated time for the object to 0 seconds. Args: @@ -830,7 +820,7 @@ def reset_time_estimate(self, **kwargs: Any) -> Dict[str, Any]: cls_names=("ProjectIssue", "ProjectMergeRequest"), required=("duration",) ) @exc.on_http_error(exc.GitlabTimeTrackingError) - def add_spent_time(self, duration: str, **kwargs: Any) -> Dict[str, Any]: + def add_spent_time(self, duration: str, **kwargs: Any) -> dict[str, Any]: """Add time spent working on the object. Args: @@ -850,7 +840,7 @@ def add_spent_time(self, duration: str, **kwargs: Any) -> Dict[str, Any]: @cli.register_custom_action(cls_names=("ProjectIssue", "ProjectMergeRequest")) @exc.on_http_error(exc.GitlabTimeTrackingError) - def reset_spent_time(self, **kwargs: Any) -> Dict[str, Any]: + def reset_spent_time(self, **kwargs: Any) -> dict[str, Any]: """Resets the time spent working on the object. Args: @@ -868,18 +858,18 @@ def reset_spent_time(self, **kwargs: Any) -> Dict[str, Any]: class ParticipantsMixin(_RestObjectBase): - _id_attr: Optional[str] - _attrs: Dict[str, Any] + _id_attr: str | None + _attrs: dict[str, Any] _module: ModuleType - _parent_attrs: Dict[str, Any] - _updated_attrs: Dict[str, Any] + _parent_attrs: dict[str, Any] + _updated_attrs: dict[str, Any] manager: base.RESTManager[Any] @cli.register_custom_action(cls_names=("ProjectMergeRequest", "ProjectIssue")) @exc.on_http_error(exc.GitlabListError) def participants( self, **kwargs: Any - ) -> Union[gitlab.client.GitlabList, List[Dict[str, Any]]]: + ) -> gitlab.client.GitlabList | list[dict[str, Any]]: """List the participants. Args: @@ -909,7 +899,7 @@ class BadgeRenderMixin(base.RESTManager[base.TObjCls]): required=("link_url", "image_url"), ) @exc.on_http_error(exc.GitlabRenderError) - def render(self, link_url: str, image_url: str, **kwargs: Any) -> Dict[str, Any]: + def render(self, link_url: str, image_url: str, **kwargs: Any) -> dict[str, Any]: """Preview link_url and image_url after interpolation. Args: @@ -933,17 +923,17 @@ def render(self, link_url: str, image_url: str, **kwargs: Any) -> Dict[str, Any] class PromoteMixin(_RestObjectBase): - _id_attr: Optional[str] - _attrs: Dict[str, Any] + _id_attr: str | None + _attrs: dict[str, Any] _module: ModuleType - _parent_attrs: Dict[str, Any] - _updated_attrs: Dict[str, Any] + _parent_attrs: dict[str, Any] + _updated_attrs: dict[str, Any] _update_method: UpdateMethod = UpdateMethod.PUT manager: base.RESTManager[Any] def _get_update_method( self, - ) -> Callable[..., Union[Dict[str, Any], requests.Response]]: + ) -> Callable[..., dict[str, Any] | requests.Response]: """Return the HTTP method to use. Returns: @@ -956,7 +946,7 @@ def _get_update_method( return http_method @exc.on_http_error(exc.GitlabPromoteError) - def promote(self, **kwargs: Any) -> Dict[str, Any]: + def promote(self, **kwargs: Any) -> dict[str, Any]: """Promote the item. Args: @@ -980,11 +970,11 @@ def promote(self, **kwargs: Any) -> Dict[str, Any]: class UploadMixin(_RestObjectBase): - _id_attr: Optional[str] - _attrs: Dict[str, Any] + _id_attr: str | None + _attrs: dict[str, Any] _module: ModuleType - _parent_attrs: Dict[str, Any] - _updated_attrs: Dict[str, Any] + _parent_attrs: dict[str, Any] + _updated_attrs: dict[str, Any] _upload_path: str manager: base.RESTManager[Any] @@ -1006,10 +996,10 @@ def _get_upload_path(self) -> str: def upload( self, filename: str, - filedata: Optional[bytes] = None, - filepath: Optional[str] = None, + filedata: bytes | None = None, + filepath: str | None = None, **kwargs: Any, - ) -> Dict[str, Any]: + ) -> dict[str, Any]: """Upload the specified file. .. note:: diff --git a/gitlab/types.py b/gitlab/types.py index 14883c6ad..e235887a7 100644 --- a/gitlab/types.py +++ b/gitlab/types.py @@ -1,18 +1,20 @@ +from __future__ import annotations + import dataclasses -from typing import Any, Dict, List, Optional, Tuple, TYPE_CHECKING +from typing import Any, TYPE_CHECKING @dataclasses.dataclass(frozen=True) class RequiredOptional: - required: Tuple[str, ...] = () - optional: Tuple[str, ...] = () - exclusive: Tuple[str, ...] = () + required: tuple[str, ...] = () + optional: tuple[str, ...] = () + exclusive: tuple[str, ...] = () def validate_attrs( self, *, - data: Dict[str, Any], - excludes: Optional[List[str]] = None, + data: dict[str, Any], + excludes: list[str] | None = None, ) -> None: if excludes is None: excludes = [] @@ -46,7 +48,7 @@ def get(self) -> Any: def set_from_cli(self, cli_value: Any) -> None: self._value = cli_value - def get_for_api(self, *, key: str) -> Tuple[str, Any]: + def get_for_api(self, *, key: str) -> tuple[str, Any]: return (key, self._value) @@ -59,7 +61,7 @@ def set_from_cli(self, cli_value: str) -> None: else: self._value = [item.strip() for item in cli_value.split(",")] - def get_for_api(self, *, key: str) -> Tuple[str, str]: + def get_for_api(self, *, key: str) -> tuple[str, str]: # Do not comma-split single value passed as string if isinstance(self._value, str): return (key, self._value) @@ -73,7 +75,7 @@ class ArrayAttribute(_ListArrayAttribute): """To support `array` types as documented in https://docs.gitlab.com/ee/api/#array""" - def get_for_api(self, *, key: str) -> Tuple[str, Any]: + def get_for_api(self, *, key: str) -> tuple[str, Any]: if isinstance(self._value, str): return (f"{key}[]", self._value) @@ -89,17 +91,17 @@ class CommaSeparatedListAttribute(_ListArrayAttribute): class LowercaseStringAttribute(GitlabAttribute): - def get_for_api(self, *, key: str) -> Tuple[str, str]: + def get_for_api(self, *, key: str) -> tuple[str, str]: return (key, str(self._value).lower()) class FileAttribute(GitlabAttribute): @staticmethod - def get_file_name(attr_name: Optional[str] = None) -> Optional[str]: + def get_file_name(attr_name: str | None = None) -> str | None: return attr_name class ImageAttribute(FileAttribute): @staticmethod - def get_file_name(attr_name: Optional[str] = None) -> str: + def get_file_name(attr_name: str | None = None) -> str: return f"{attr_name}.png" if attr_name else "image.png" diff --git a/gitlab/utils.py b/gitlab/utils.py index d26518b3e..d804c404d 100644 --- a/gitlab/utils.py +++ b/gitlab/utils.py @@ -1,3 +1,5 @@ +from __future__ import annotations + import dataclasses import email.message import logging @@ -6,17 +8,11 @@ import traceback import urllib.parse import warnings +from collections.abc import Iterator, MutableMapping from typing import ( Any, Callable, - Dict, - Iterator, Literal, - MutableMapping, - Optional, - Tuple, - Type, - Union, ) import requests @@ -29,7 +25,7 @@ def __call__(self, chunk: Any) -> None: print(chunk) -def get_base_url(url: Optional[str] = None) -> str: +def get_base_url(url: str | None = None) -> str: """Return the base URL with the trailing slash stripped. If the URL is a Falsy value, return the default URL. Returns: @@ -41,7 +37,7 @@ def get_base_url(url: Optional[str] = None) -> str: return url.rstrip("/") -def get_content_type(content_type: Optional[str]) -> str: +def get_content_type(content_type: str | None) -> str: message = email.message.Message() if content_type is not None: message["content-type"] = content_type @@ -54,11 +50,11 @@ class MaskingFormatter(logging.Formatter): def __init__( self, - fmt: Optional[str] = logging.BASIC_FORMAT, - datefmt: Optional[str] = None, + fmt: str | None = logging.BASIC_FORMAT, + datefmt: str | None = None, style: Literal["%", "{", "$"] = "%", validate: bool = True, - masked: Optional[str] = None, + masked: str | None = None, ) -> None: super().__init__(fmt, datefmt, style, validate) self.masked = masked @@ -77,11 +73,11 @@ def format(self, record: logging.LogRecord) -> str: def response_content( response: requests.Response, streamed: bool, - action: Optional[Callable[[bytes], Any]], + action: Callable[[bytes], Any] | None, chunk_size: int, *, iterator: bool, -) -> Optional[Union[bytes, Iterator[Any]]]: +) -> bytes | Iterator[Any] | None: if iterator: return response.iter_content(chunk_size=chunk_size) @@ -101,17 +97,15 @@ class Retry: def __init__( self, max_retries: int, - obey_rate_limit: Optional[bool] = True, - retry_transient_errors: Optional[bool] = False, + obey_rate_limit: bool | None = True, + retry_transient_errors: bool | None = False, ) -> None: self.cur_retries = 0 self.max_retries = max_retries self.obey_rate_limit = obey_rate_limit self.retry_transient_errors = retry_transient_errors - def _retryable_status_code( - self, status_code: Optional[int], reason: str = "" - ) -> bool: + def _retryable_status_code(self, status_code: int | None, reason: str = "") -> bool: if status_code == 429 and self.obey_rate_limit: return True @@ -126,8 +120,8 @@ def _retryable_status_code( def handle_retry_on_status( self, - status_code: Optional[int], - headers: Optional[MutableMapping[str, str]] = None, + status_code: int | None, + headers: MutableMapping[str, str] | None = None, reason: str = "", ) -> bool: if not self._retryable_status_code(status_code, reason): @@ -163,12 +157,12 @@ def handle_retry(self) -> bool: def _transform_types( - data: Dict[str, Any], - custom_types: Dict[str, Any], + data: dict[str, Any], + custom_types: dict[str, Any], *, transform_data: bool, - transform_files: Optional[bool] = True, -) -> Tuple[Dict[str, Any], Dict[str, Any]]: + transform_files: bool | None = True, +) -> tuple[dict[str, Any], dict[str, Any]]: """Copy the data dict with attributes that have custom types and transform them before being sent to the server. @@ -216,8 +210,8 @@ def _transform_types( def copy_dict( *, - src: Dict[str, Any], - dest: Dict[str, Any], + src: dict[str, Any], + dest: dict[str, Any], ) -> None: for k, v in src.items(): if isinstance(v, dict): @@ -247,7 +241,7 @@ class EncodedId(str): https://docs.gitlab.com/ee/api/index.html#path-parameters """ - def __new__(cls, value: Union[str, int, "EncodedId"]) -> "EncodedId": + def __new__(cls, value: str | int | EncodedId) -> EncodedId: if isinstance(value, EncodedId): return value @@ -258,15 +252,15 @@ def __new__(cls, value: Union[str, int, "EncodedId"]) -> "EncodedId": return super().__new__(cls, value) -def remove_none_from_dict(data: Dict[str, Any]) -> Dict[str, Any]: +def remove_none_from_dict(data: dict[str, Any]) -> dict[str, Any]: return {k: v for k, v in data.items() if v is not None} def warn( message: str, *, - category: Optional[Type[Warning]] = None, - source: Optional[Any] = None, + category: type[Warning] | None = None, + source: Any | None = None, show_caller: bool = True, ) -> None: """This `warnings.warn` wrapper function attempts to show the location causing the diff --git a/gitlab/v4/cli.py b/gitlab/v4/cli.py index a7a7fb6b6..9727728c1 100644 --- a/gitlab/v4/cli.py +++ b/gitlab/v4/cli.py @@ -1,8 +1,10 @@ +from __future__ import annotations + import argparse import json import operator import sys -from typing import Any, Dict, List, Optional, Type, TYPE_CHECKING, Union +from typing import Any, TYPE_CHECKING import gitlab import gitlab.base @@ -17,9 +19,9 @@ def __init__( gl: gitlab.Gitlab, gitlab_resource: str, resource_action: str, - args: Dict[str, str], + args: dict[str, str], ) -> None: - self.cls: Type[gitlab.base.RESTObject] = cli.gitlab_resource_to_cls( + self.cls: type[gitlab.base.RESTObject] = cli.gitlab_resource_to_cls( gitlab_resource, namespace=gitlab.v4.objects ) self.cls_name = self.cls.__name__ @@ -27,7 +29,7 @@ def __init__( self.resource_action = resource_action.lower() self.gl = gl self.args = args - self.parent_args: Dict[str, Any] = {} + self.parent_args: dict[str, Any] = {} self.mgr_cls: Any = getattr(gitlab.v4.objects, f"{self.cls.__name__}Manager") # We could do something smart, like splitting the manager name to find # parents, build the chain of managers to get to the final object. @@ -73,9 +75,9 @@ def run(self) -> Any: return self.do_custom() def do_custom(self) -> Any: - class_instance: Union[ - gitlab.base.RESTManager[gitlab.base.RESTObject], gitlab.base.RESTObject - ] + class_instance: ( + gitlab.base.RESTManager[gitlab.base.RESTObject] | gitlab.base.RESTObject + ) in_obj = cli.custom_actions[self.cls_name][self.resource_action].in_object # Get the object (lazy), then act @@ -133,7 +135,7 @@ def do_create(self) -> gitlab.base.RESTObject: def do_list( self, - ) -> Union[gitlab.base.RESTObjectList, List[gitlab.base.RESTObject]]: + ) -> gitlab.base.RESTObjectList | list[gitlab.base.RESTObject]: if TYPE_CHECKING: assert isinstance(self.mgr, gitlab.mixins.ListMixin) message_details = gitlab.utils.WarnMessageData( @@ -150,7 +152,7 @@ def do_list( cli.die("Impossible to list objects", e) return result - def do_get(self) -> Optional[gitlab.base.RESTObject]: + def do_get(self) -> gitlab.base.RESTObject | None: if isinstance(self.mgr, gitlab.mixins.GetWithoutIdMixin): try: result = self.mgr.get(id=None, **self.args) @@ -183,7 +185,7 @@ def do_delete(self) -> None: except Exception as e: # pragma: no cover, cli.die is unit-tested cli.die("Impossible to destroy object", e) - def do_update(self) -> Dict[str, Any]: + def do_update(self) -> dict[str, Any]: if TYPE_CHECKING: assert isinstance(self.mgr, gitlab.mixins.UpdateMixin) if issubclass(self.mgr_cls, gitlab.mixins.GetWithoutIdMixin): @@ -208,13 +210,13 @@ def do_update(self) -> Dict[str, Any]: def _populate_sub_parser_by_class( - cls: Type[gitlab.base.RESTObject], + cls: type[gitlab.base.RESTObject], sub_parser: _SubparserType, ) -> None: mgr_cls_name = f"{cls.__name__}Manager" mgr_cls = getattr(gitlab.v4.objects, mgr_cls_name) - action_parsers: Dict[str, argparse.ArgumentParser] = {} + action_parsers: dict[str, argparse.ArgumentParser] = {} for action_name, help_text in [ ("list", "List the GitLab resources"), ("get", "Get a GitLab resource"), @@ -429,8 +431,8 @@ def extend_parser(parser: argparse.ArgumentParser) -> argparse.ArgumentParser: def get_dict( - obj: Union[str, Dict[str, Any], gitlab.base.RESTObject], fields: List[str] -) -> Union[str, Dict[str, Any]]: + obj: str | dict[str, Any] | gitlab.base.RESTObject, fields: list[str] +) -> str | dict[str, Any]: if not isinstance(obj, gitlab.base.RESTObject): return obj @@ -441,13 +443,13 @@ def get_dict( class JSONPrinter: @staticmethod - def display(d: Union[str, Dict[str, Any]], **_kwargs: Any) -> None: + def display(d: str | dict[str, Any], **_kwargs: Any) -> None: print(json.dumps(d)) @staticmethod def display_list( - data: List[Union[str, Dict[str, Any], gitlab.base.RESTObject]], - fields: List[str], + data: list[str | dict[str, Any] | gitlab.base.RESTObject], + fields: list[str], **_kwargs: Any, ) -> None: print(json.dumps([get_dict(obj, fields) for obj in data])) @@ -455,7 +457,7 @@ def display_list( class YAMLPrinter: @staticmethod - def display(d: Union[str, Dict[str, Any]], **_kwargs: Any) -> None: + def display(d: str | dict[str, Any], **_kwargs: Any) -> None: try: import yaml # noqa @@ -469,8 +471,8 @@ def display(d: Union[str, Dict[str, Any]], **_kwargs: Any) -> None: @staticmethod def display_list( - data: List[Union[str, Dict[str, Any], gitlab.base.RESTObject]], - fields: List[str], + data: list[str | dict[str, Any] | gitlab.base.RESTObject], + fields: list[str], **_kwargs: Any, ) -> None: try: @@ -490,14 +492,14 @@ def display_list( class LegacyPrinter: - def display(self, _d: Union[str, Dict[str, Any]], **kwargs: Any) -> None: + def display(self, _d: str | dict[str, Any], **kwargs: Any) -> None: verbose = kwargs.get("verbose", False) padding = kwargs.get("padding", 0) - obj: Optional[Union[Dict[str, Any], gitlab.base.RESTObject]] = kwargs.get("obj") + obj: dict[str, Any] | gitlab.base.RESTObject | None = kwargs.get("obj") if TYPE_CHECKING: assert obj is not None - def display_dict(d: Dict[str, Any], padding: int) -> None: + def display_dict(d: dict[str, Any], padding: int) -> None: for k in sorted(d.keys()): v = d[k] if isinstance(v, dict): @@ -551,10 +553,7 @@ def display_dict(d: Dict[str, Any], padding: int) -> None: ) def display_list( - self, - data: List[Union[str, gitlab.base.RESTObject]], - fields: List[str], - **kwargs: Any, + self, data: list[str | gitlab.base.RESTObject], fields: list[str], **kwargs: Any ) -> None: verbose = kwargs.get("verbose", False) for obj in data: @@ -565,9 +564,7 @@ def display_list( print("") -PRINTERS: Dict[ - str, Union[Type[JSONPrinter], Type[LegacyPrinter], Type[YAMLPrinter]] -] = { +PRINTERS: dict[str, type[JSONPrinter] | type[LegacyPrinter] | type[YAMLPrinter]] = { "json": JSONPrinter, "legacy": LegacyPrinter, "yaml": YAMLPrinter, @@ -578,10 +575,10 @@ def run( gl: gitlab.Gitlab, gitlab_resource: str, resource_action: str, - args: Dict[str, Any], + args: dict[str, Any], verbose: bool, output: str, - fields: List[str], + fields: list[str], ) -> None: g_cli = GitlabCLI( gl=gl, @@ -591,7 +588,7 @@ def run( ) data = g_cli.run() - printer: Union[JSONPrinter, LegacyPrinter, YAMLPrinter] = PRINTERS[output]() + printer: JSONPrinter | LegacyPrinter | YAMLPrinter = PRINTERS[output]() if isinstance(data, dict): printer.display(data, verbose=True, obj=data) diff --git a/gitlab/v4/objects/appearance.py b/gitlab/v4/objects/appearance.py index cdc99ad2c..f894120c4 100644 --- a/gitlab/v4/objects/appearance.py +++ b/gitlab/v4/objects/appearance.py @@ -1,4 +1,6 @@ -from typing import Any, Dict, Optional, Union +from __future__ import annotations + +from typing import Any from gitlab import exceptions as exc from gitlab.base import RESTObject @@ -39,10 +41,10 @@ class ApplicationAppearanceManager( @exc.on_http_error(exc.GitlabUpdateError) def update( self, - id: Optional[Union[str, int]] = None, - new_data: Optional[Dict[str, Any]] = None, + id: str | int | None = None, + new_data: dict[str, Any] | None = None, **kwargs: Any, - ) -> Dict[str, Any]: + ) -> dict[str, Any]: """Update an object on the server. Args: diff --git a/gitlab/v4/objects/artifacts.py b/gitlab/v4/objects/artifacts.py index 45a40322b..2007ff4af 100644 --- a/gitlab/v4/objects/artifacts.py +++ b/gitlab/v4/objects/artifacts.py @@ -3,15 +3,15 @@ https://docs.gitlab.com/ee/api/job_artifacts.html """ +from __future__ import annotations + from typing import ( Any, Callable, Iterator, Literal, - Optional, overload, TYPE_CHECKING, - Union, ) import requests @@ -84,7 +84,7 @@ def download( ref_name: str, job: str, streamed: Literal[True] = True, - action: Optional[Callable[[bytes], Any]] = None, + action: Callable[[bytes], Any] | None = None, chunk_size: int = 1024, *, iterator: Literal[False] = False, @@ -102,12 +102,12 @@ def download( ref_name: str, job: str, streamed: bool = False, - action: Optional[Callable[[bytes], Any]] = None, + action: Callable[[bytes], Any] | None = None, chunk_size: int = 1024, *, iterator: bool = False, **kwargs: Any, - ) -> Optional[Union[bytes, Iterator[Any]]]: + ) -> bytes | Iterator[Any] | None: """Get the job artifacts archive from a specific tag or branch. Args: @@ -177,7 +177,7 @@ def raw( artifact_path: str, job: str, streamed: Literal[True] = True, - action: Optional[Callable[[bytes], Any]] = None, + action: Callable[[bytes], Any] | None = None, chunk_size: int = 1024, *, iterator: Literal[False] = False, @@ -195,12 +195,12 @@ def raw( artifact_path: str, job: str, streamed: bool = False, - action: Optional[Callable[[bytes], Any]] = None, + action: Callable[[bytes], Any] | None = None, chunk_size: int = 1024, *, iterator: bool = False, **kwargs: Any, - ) -> Optional[Union[bytes, Iterator[Any]]]: + ) -> bytes | Iterator[Any] | None: """Download a single artifact file from a specific tag or branch from within the job's artifacts archive. diff --git a/gitlab/v4/objects/bulk_imports.py b/gitlab/v4/objects/bulk_imports.py index a08a228ae..b171618a5 100644 --- a/gitlab/v4/objects/bulk_imports.py +++ b/gitlab/v4/objects/bulk_imports.py @@ -1,3 +1,5 @@ +from __future__ import annotations + from gitlab.base import RESTObject from gitlab.mixins import CreateMixin, ListMixin, RefreshMixin, RetrieveMixin from gitlab.types import RequiredOptional @@ -13,7 +15,7 @@ class BulkImport(RefreshMixin, RESTObject): - entities: "BulkImportEntityManager" + entities: BulkImportEntityManager class BulkImportManager(CreateMixin[BulkImport], RetrieveMixin[BulkImport]): diff --git a/gitlab/v4/objects/clusters.py b/gitlab/v4/objects/clusters.py index 4f3f2d5a1..7632a55a0 100644 --- a/gitlab/v4/objects/clusters.py +++ b/gitlab/v4/objects/clusters.py @@ -1,4 +1,6 @@ -from typing import Any, Dict, Optional +from __future__ import annotations + +from typing import Any from gitlab import exceptions as exc from gitlab.base import RESTObject @@ -36,9 +38,7 @@ class GroupClusterManager(CRUDMixin[GroupCluster]): ) @exc.on_http_error(exc.GitlabStopError) - def create( - self, data: Optional[Dict[str, Any]] = None, **kwargs: Any - ) -> GroupCluster: + def create(self, data: dict[str, Any] | None = None, **kwargs: Any) -> GroupCluster: """Create a new object. Args: @@ -83,7 +83,7 @@ class ProjectClusterManager(CRUDMixin[ProjectCluster]): @exc.on_http_error(exc.GitlabStopError) def create( - self, data: Optional[Dict[str, Any]] = None, **kwargs: Any + self, data: dict[str, Any] | None = None, **kwargs: Any ) -> ProjectCluster: """Create a new object. diff --git a/gitlab/v4/objects/commits.py b/gitlab/v4/objects/commits.py index 7957922b4..54402e278 100644 --- a/gitlab/v4/objects/commits.py +++ b/gitlab/v4/objects/commits.py @@ -1,4 +1,6 @@ -from typing import Any, Dict, List, Optional, TYPE_CHECKING, Union +from __future__ import annotations + +from typing import Any, TYPE_CHECKING import requests @@ -24,13 +26,13 @@ class ProjectCommit(RESTObject): _repr_attr = "title" - comments: "ProjectCommitCommentManager" + comments: ProjectCommitCommentManager discussions: ProjectCommitDiscussionManager - statuses: "ProjectCommitStatusManager" + statuses: ProjectCommitStatusManager @cli.register_custom_action(cls_names="ProjectCommit") @exc.on_http_error(exc.GitlabGetError) - def diff(self, **kwargs: Any) -> Union[gitlab.GitlabList, List[Dict[str, Any]]]: + def diff(self, **kwargs: Any) -> gitlab.GitlabList | list[dict[str, Any]]: """Generate the commit diff. Args: @@ -50,7 +52,7 @@ def diff(self, **kwargs: Any) -> Union[gitlab.GitlabList, List[Dict[str, Any]]]: @exc.on_http_error(exc.GitlabCherryPickError) def cherry_pick( self, branch: str, **kwargs: Any - ) -> Union[Dict[str, Any], requests.Response]: + ) -> dict[str, Any] | requests.Response: """Cherry-pick a commit into a branch. Args: @@ -72,7 +74,7 @@ def cherry_pick( @exc.on_http_error(exc.GitlabGetError) def refs( self, type: str = "all", **kwargs: Any - ) -> Union[gitlab.GitlabList, List[Dict[str, Any]]]: + ) -> gitlab.GitlabList | list[dict[str, Any]]: """List the references the commit is pushed to. Args: @@ -92,9 +94,7 @@ def refs( @cli.register_custom_action(cls_names="ProjectCommit") @exc.on_http_error(exc.GitlabGetError) - def merge_requests( - self, **kwargs: Any - ) -> Union[gitlab.GitlabList, List[Dict[str, Any]]]: + def merge_requests(self, **kwargs: Any) -> gitlab.GitlabList | list[dict[str, Any]]: """List the merge requests related to the commit. Args: @@ -112,9 +112,7 @@ def merge_requests( @cli.register_custom_action(cls_names="ProjectCommit", required=("branch",)) @exc.on_http_error(exc.GitlabRevertError) - def revert( - self, branch: str, **kwargs: Any - ) -> Union[Dict[str, Any], requests.Response]: + def revert(self, branch: str, **kwargs: Any) -> dict[str, Any] | requests.Response: """Revert a commit on a given branch. Args: @@ -134,7 +132,7 @@ def revert( @cli.register_custom_action(cls_names="ProjectCommit") @exc.on_http_error(exc.GitlabGetError) - def sequence(self, **kwargs: Any) -> Union[Dict[str, Any], requests.Response]: + def sequence(self, **kwargs: Any) -> dict[str, Any] | requests.Response: """Get the sequence number of the commit. Args: @@ -152,7 +150,7 @@ def sequence(self, **kwargs: Any) -> Union[Dict[str, Any], requests.Response]: @cli.register_custom_action(cls_names="ProjectCommit") @exc.on_http_error(exc.GitlabGetError) - def signature(self, **kwargs: Any) -> Union[Dict[str, Any], requests.Response]: + def signature(self, **kwargs: Any) -> dict[str, Any] | requests.Response: """Get the signature of the commit. Args: @@ -223,7 +221,7 @@ class ProjectCommitStatusManager( @exc.on_http_error(exc.GitlabCreateError) def create( - self, data: Optional[Dict[str, Any]] = None, **kwargs: Any + self, data: dict[str, Any] | None = None, **kwargs: Any ) -> ProjectCommitStatus: """Create a new object. @@ -245,7 +243,7 @@ def create( # they are missing when using only the API # See #511 base_path = "/projects/{project_id}/statuses/{commit_id}" - path: Optional[str] + path: str | None if data is not None and "project_id" in data and "commit_id" in data: path = base_path.format(**data) else: diff --git a/gitlab/v4/objects/container_registry.py b/gitlab/v4/objects/container_registry.py index cb7904ee4..c8165126b 100644 --- a/gitlab/v4/objects/container_registry.py +++ b/gitlab/v4/objects/container_registry.py @@ -1,3 +1,5 @@ +from __future__ import annotations + from typing import Any from gitlab import cli @@ -23,7 +25,7 @@ class ProjectRegistryRepository(ObjectDeleteMixin, RESTObject): - tags: "ProjectRegistryTagManager" + tags: ProjectRegistryTagManager class ProjectRegistryRepositoryManager( diff --git a/gitlab/v4/objects/deploy_keys.py b/gitlab/v4/objects/deploy_keys.py index c959357b4..40ba71f8d 100644 --- a/gitlab/v4/objects/deploy_keys.py +++ b/gitlab/v4/objects/deploy_keys.py @@ -1,4 +1,6 @@ -from typing import Any, Dict, Union +from __future__ import annotations + +from typing import Any import requests @@ -43,9 +45,7 @@ class ProjectKeyManager(CRUDMixin[ProjectKey]): help="Enable a deploy key for the project", ) @exc.on_http_error(exc.GitlabProjectDeployKeyError) - def enable( - self, key_id: int, **kwargs: Any - ) -> Union[Dict[str, Any], requests.Response]: + def enable(self, key_id: int, **kwargs: Any) -> dict[str, Any] | requests.Response: """Enable a deploy key for a project. Args: diff --git a/gitlab/v4/objects/deployments.py b/gitlab/v4/objects/deployments.py index 2c0de96cd..03d67338a 100644 --- a/gitlab/v4/objects/deployments.py +++ b/gitlab/v4/objects/deployments.py @@ -3,7 +3,9 @@ https://docs.gitlab.com/ee/api/deployments.html """ -from typing import Any, Dict, Optional, TYPE_CHECKING +from __future__ import annotations + +from typing import Any, TYPE_CHECKING from gitlab import cli from gitlab import exceptions as exc @@ -31,10 +33,10 @@ class ProjectDeployment(SaveMixin, RESTObject): def approval( self, status: str, - comment: Optional[str] = None, - represented_as: Optional[str] = None, + comment: str | None = None, + represented_as: str | None = None, **kwargs: Any, - ) -> Dict[str, Any]: + ) -> dict[str, Any]: """Approve or reject a blocked deployment. Args: diff --git a/gitlab/v4/objects/environments.py b/gitlab/v4/objects/environments.py index b69f3d713..87d62f506 100644 --- a/gitlab/v4/objects/environments.py +++ b/gitlab/v4/objects/environments.py @@ -1,4 +1,6 @@ -from typing import Any, Dict, Union +from __future__ import annotations + +from typing import Any import requests @@ -26,7 +28,7 @@ class ProjectEnvironment(SaveMixin, ObjectDeleteMixin, RESTObject): @cli.register_custom_action(cls_names="ProjectEnvironment") @exc.on_http_error(exc.GitlabStopError) - def stop(self, **kwargs: Any) -> Union[Dict[str, Any], requests.Response]: + def stop(self, **kwargs: Any) -> dict[str, Any] | requests.Response: """Stop the environment. Args: diff --git a/gitlab/v4/objects/epics.py b/gitlab/v4/objects/epics.py index b882d57f3..541a63875 100644 --- a/gitlab/v4/objects/epics.py +++ b/gitlab/v4/objects/epics.py @@ -1,4 +1,6 @@ -from typing import Any, Dict, Optional, TYPE_CHECKING +from __future__ import annotations + +from typing import Any, TYPE_CHECKING from gitlab import exceptions as exc from gitlab import types @@ -28,7 +30,7 @@ class GroupEpic(ObjectDeleteMixin, SaveMixin, RESTObject): _id_attr = "iid" - issues: "GroupEpicIssueManager" + issues: GroupEpicIssueManager resourcelabelevents: GroupEpicResourceLabelEventManager notes: GroupEpicNoteManager @@ -52,7 +54,7 @@ class GroupEpicIssue(ObjectDeleteMixin, SaveMixin, RESTObject): _id_attr = "epic_issue_id" # Define type for 'manager' here So mypy won't complain about # 'self.manager.update()' call in the 'save' method. - manager: "GroupEpicIssueManager" + manager: GroupEpicIssueManager def save(self, **kwargs: Any) -> None: """Save the changes made to the object to the server. @@ -90,7 +92,7 @@ class GroupEpicIssueManager( @exc.on_http_error(exc.GitlabCreateError) def create( - self, data: Optional[Dict[str, Any]] = None, **kwargs: Any + self, data: dict[str, Any] | None = None, **kwargs: Any ) -> GroupEpicIssue: """Create a new object. diff --git a/gitlab/v4/objects/features.py b/gitlab/v4/objects/features.py index f6a76e910..806b2e267 100644 --- a/gitlab/v4/objects/features.py +++ b/gitlab/v4/objects/features.py @@ -3,7 +3,9 @@ https://docs.gitlab.com/ee/api/features.html """ -from typing import Any, Optional, TYPE_CHECKING, Union +from __future__ import annotations + +from typing import Any, TYPE_CHECKING from gitlab import exceptions as exc from gitlab import utils @@ -28,11 +30,11 @@ class FeatureManager(ListMixin[Feature], DeleteMixin[Feature]): def set( self, name: str, - value: Union[bool, int], - feature_group: Optional[str] = None, - user: Optional[str] = None, - group: Optional[str] = None, - project: Optional[str] = None, + value: bool | int, + feature_group: str | None = None, + user: str | None = None, + group: str | None = None, + project: str | None = None, **kwargs: Any, ) -> Feature: """Create or update the object. diff --git a/gitlab/v4/objects/files.py b/gitlab/v4/objects/files.py index 690946b52..d2ed6e4fa 100644 --- a/gitlab/v4/objects/files.py +++ b/gitlab/v4/objects/files.py @@ -1,16 +1,13 @@ +from __future__ import annotations + import base64 from typing import ( Any, Callable, - Dict, Iterator, - List, Literal, - Optional, overload, - Tuple, TYPE_CHECKING, - Union, ) import requests @@ -40,7 +37,7 @@ class ProjectFile(SaveMixin, ObjectDeleteMixin, RESTObject): branch: str commit_message: str file_path: str - manager: "ProjectFileManager" + manager: ProjectFileManager content: str # since the `decode()` method uses `self.content` def decode(self) -> bytes: @@ -103,7 +100,7 @@ class ProjectFileManager( _path = "/projects/{project_id}/repository/files" _obj_cls = ProjectFile _from_parent_attrs = {"project_id": "id"} - _optional_get_attrs: Tuple[str, ...] = () + _optional_get_attrs: tuple[str, ...] = () _create_attrs = RequiredOptional( required=("file_path", "branch", "content", "commit_message"), optional=( @@ -157,7 +154,7 @@ def get(self, file_path: str, ref: str, **kwargs: Any) -> ProjectFile: @exc.on_http_error(exc.GitlabHeadError) def head( self, file_path: str, ref: str, **kwargs: Any - ) -> "requests.structures.CaseInsensitiveDict[Any]": + ) -> requests.structures.CaseInsensitiveDict[Any]: """Retrieve just metadata for a single file. Args: @@ -190,9 +187,7 @@ def head( ), ) @exc.on_http_error(exc.GitlabCreateError) - def create( - self, data: Optional[Dict[str, Any]] = None, **kwargs: Any - ) -> ProjectFile: + def create(self, data: dict[str, Any] | None = None, **kwargs: Any) -> ProjectFile: """Create a new object. Args: @@ -224,8 +219,8 @@ def create( # NOTE(jlvillal): Signature doesn't match UpdateMixin.update() so ignore # type error def update( # type: ignore[override] - self, file_path: str, new_data: Optional[Dict[str, Any]] = None, **kwargs: Any - ) -> Dict[str, Any]: + self, file_path: str, new_data: dict[str, Any] | None = None, **kwargs: Any + ) -> dict[str, Any]: """Update an object on the server. Args: @@ -282,7 +277,7 @@ def delete( # type: ignore[override] def raw( self, file_path: str, - ref: Optional[str] = None, + ref: str | None = None, streamed: Literal[False] = False, action: None = None, chunk_size: int = 1024, @@ -295,7 +290,7 @@ def raw( def raw( self, file_path: str, - ref: Optional[str] = None, + ref: str | None = None, streamed: bool = False, action: None = None, chunk_size: int = 1024, @@ -308,9 +303,9 @@ def raw( def raw( self, file_path: str, - ref: Optional[str] = None, + ref: str | None = None, streamed: Literal[True] = True, - action: Optional[Callable[[bytes], Any]] = None, + action: Callable[[bytes], Any] | None = None, chunk_size: int = 1024, *, iterator: Literal[False] = False, @@ -326,14 +321,14 @@ def raw( def raw( self, file_path: str, - ref: Optional[str] = None, + ref: str | None = None, streamed: bool = False, - action: Optional[Callable[..., Any]] = None, + action: Callable[..., Any] | None = None, chunk_size: int = 1024, *, iterator: bool = False, **kwargs: Any, - ) -> Optional[Union[bytes, Iterator[Any]]]: + ) -> bytes | Iterator[Any] | None: """Return the content of a file for a commit. Args: @@ -375,7 +370,7 @@ def raw( cls_names="ProjectFileManager", required=("file_path", "ref") ) @exc.on_http_error(exc.GitlabListError) - def blame(self, file_path: str, ref: str, **kwargs: Any) -> List[Dict[str, Any]]: + def blame(self, file_path: str, ref: str, **kwargs: Any) -> list[dict[str, Any]]: """Return the content of a file for a commit. Args: diff --git a/gitlab/v4/objects/groups.py b/gitlab/v4/objects/groups.py index 82f09632e..43ba5564c 100644 --- a/gitlab/v4/objects/groups.py +++ b/gitlab/v4/objects/groups.py @@ -1,4 +1,6 @@ -from typing import Any, BinaryIO, Dict, List, Optional, TYPE_CHECKING, Union +from __future__ import annotations + +from typing import Any, BinaryIO, TYPE_CHECKING import requests @@ -79,7 +81,7 @@ class Group(SaveMixin, ObjectDeleteMixin, RESTObject): clusters: GroupClusterManager customattributes: GroupCustomAttributeManager deploytokens: GroupDeployTokenManager - descendant_groups: "GroupDescendantGroupManager" + descendant_groups: GroupDescendantGroupManager epics: GroupEpicManager exports: GroupExportManager hooks: GroupHookManager @@ -89,7 +91,7 @@ class Group(SaveMixin, ObjectDeleteMixin, RESTObject): issues_statistics: GroupIssuesStatisticsManager iterations: GroupIterationManager labels: GroupLabelManager - ldap_group_links: "GroupLDAPGroupLinkManager" + ldap_group_links: GroupLDAPGroupLinkManager members: GroupMemberManager members_all: GroupMemberAllManager mergerequests: GroupMergeRequestManager @@ -101,11 +103,11 @@ class Group(SaveMixin, ObjectDeleteMixin, RESTObject): pushrules: GroupPushRulesManager registry_repositories: GroupRegistryRepositoryManager runners: GroupRunnerManager - subgroups: "GroupSubgroupManager" + subgroups: GroupSubgroupManager variables: GroupVariableManager wikis: GroupWikiManager - saml_group_links: "GroupSAMLGroupLinkManager" - service_accounts: "GroupServiceAccountManager" + saml_group_links: GroupSAMLGroupLinkManager + service_accounts: GroupServiceAccountManager @cli.register_custom_action(cls_names="Group", required=("project_id",)) @exc.on_http_error(exc.GitlabTransferProjectError) @@ -125,7 +127,7 @@ def transfer_project(self, project_id: int, **kwargs: Any) -> None: @cli.register_custom_action(cls_names="Group", required=(), optional=("group_id",)) @exc.on_http_error(exc.GitlabGroupTransferError) - def transfer(self, group_id: Optional[int] = None, **kwargs: Any) -> None: + def transfer(self, group_id: int | None = None, **kwargs: Any) -> None: """Transfer the group to a new parent group or make it a top-level group. Requires GitLab ≥14.6. @@ -149,7 +151,7 @@ def transfer(self, group_id: Optional[int] = None, **kwargs: Any) -> None: @exc.on_http_error(exc.GitlabSearchError) def search( self, scope: str, search: str, **kwargs: Any - ) -> Union[gitlab.GitlabList, List[Dict[str, Any]]]: + ) -> gitlab.GitlabList | list[dict[str, Any]]: """Search the group resources matching the provided string. Args: @@ -193,7 +195,7 @@ def share( self, group_id: int, group_access: int, - expires_at: Optional[str] = None, + expires_at: str | None = None, **kwargs: Any, ) -> None: """Share the group with a group. @@ -325,9 +327,9 @@ def import_group( file: BinaryIO, path: str, name: str, - parent_id: Optional[Union[int, str]] = None, + parent_id: int | str | None = None, **kwargs: Any, - ) -> Union[Dict[str, Any], requests.Response]: + ) -> dict[str, Any] | requests.Response: """Import a group from an archive file. Args: @@ -346,7 +348,7 @@ def import_group( A representation of the import status. """ files = {"file": ("file.tar.gz", file, "application/octet-stream")} - data: Dict[str, Any] = {"path": path, "name": name} + data: dict[str, Any] = {"path": path, "name": name} if parent_id is not None: data["parent_id"] = parent_id @@ -397,7 +399,7 @@ class GroupDescendantGroupManager(SubgroupBaseManager[GroupDescendantGroup]): class GroupLDAPGroupLink(RESTObject): _repr_attr = "provider" - def _get_link_attrs(self) -> Dict[str, str]: + def _get_link_attrs(self) -> dict[str, str]: # https://docs.gitlab.com/ee/api/groups.html#add-ldap-group-link-with-cn-or-filter # https://docs.gitlab.com/ee/api/groups.html#delete-ldap-group-link-with-cn-or-filter # We can tell what attribute to use based on the data returned diff --git a/gitlab/v4/objects/invitations.py b/gitlab/v4/objects/invitations.py index 789a77e34..5eb4c645c 100644 --- a/gitlab/v4/objects/invitations.py +++ b/gitlab/v4/objects/invitations.py @@ -1,4 +1,6 @@ -from typing import Any, Dict, Optional +from __future__ import annotations + +from typing import Any from gitlab.base import RESTObject, TObjCls from gitlab.exceptions import GitlabInvitationError @@ -15,11 +17,7 @@ class InvitationMixin(CRUDMixin[TObjCls]): # pylint: disable=abstract-method - def create( - self, - data: Optional[Dict[str, Any]] = None, - **kwargs: Any, - ) -> TObjCls: + def create(self, data: dict[str, Any] | None = None, **kwargs: Any) -> TObjCls: invitation = super().create(data, **kwargs) if invitation.status == "error": diff --git a/gitlab/v4/objects/issues.py b/gitlab/v4/objects/issues.py index c99251bd0..46fe762ff 100644 --- a/gitlab/v4/objects/issues.py +++ b/gitlab/v4/objects/issues.py @@ -1,4 +1,6 @@ -from typing import Any, Dict, List, Optional, Tuple, TYPE_CHECKING, Union +from __future__ import annotations + +from typing import Any, TYPE_CHECKING import requests @@ -117,7 +119,7 @@ class ProjectIssue( awardemojis: ProjectIssueAwardEmojiManager discussions: ProjectIssueDiscussionManager - links: "ProjectIssueLinkManager" + links: ProjectIssueLinkManager notes: ProjectIssueNoteManager resourcelabelevents: ProjectIssueResourceLabelEventManager resourcemilestoneevents: ProjectIssueResourceMilestoneEventManager @@ -151,8 +153,8 @@ def move(self, to_project_id: int, **kwargs: Any) -> None: @exc.on_http_error(exc.GitlabUpdateError) def reorder( self, - move_after_id: Optional[int] = None, - move_before_id: Optional[int] = None, + move_after_id: int | None = None, + move_before_id: int | None = None, **kwargs: Any, ) -> None: """Reorder an issue on a board. @@ -167,7 +169,7 @@ def reorder( GitlabUpdateError: If the issue could not be reordered """ path = f"{self.manager.path}/{self.encoded_id}/reorder" - data: Dict[str, Any] = {} + data: dict[str, Any] = {} if move_after_id is not None: data["move_after_id"] = move_after_id @@ -183,7 +185,7 @@ def reorder( @exc.on_http_error(exc.GitlabGetError) def related_merge_requests( self, **kwargs: Any - ) -> Union[client.GitlabList, List[Dict[str, Any]]]: + ) -> client.GitlabList | list[dict[str, Any]]: """List merge requests related to the issue. Args: @@ -204,9 +206,7 @@ def related_merge_requests( @cli.register_custom_action(cls_names="ProjectIssue") @exc.on_http_error(exc.GitlabGetError) - def closed_by( - self, **kwargs: Any - ) -> Union[client.GitlabList, List[Dict[str, Any]]]: + def closed_by(self, **kwargs: Any) -> client.GitlabList | list[dict[str, Any]]: """List merge requests that will close the issue when merged. Args: @@ -299,8 +299,8 @@ class ProjectIssueLinkManager( # NOTE(jlvillal): Signature doesn't match CreateMixin.create() so ignore # type error def create( # type: ignore[override] - self, data: Dict[str, Any], **kwargs: Any - ) -> Tuple[ProjectIssue, ProjectIssue]: + self, data: dict[str, Any], **kwargs: Any + ) -> tuple[ProjectIssue, ProjectIssue]: """Create a new object. Args: diff --git a/gitlab/v4/objects/job_token_scope.py b/gitlab/v4/objects/job_token_scope.py index b6bd8f9a0..b780298ed 100644 --- a/gitlab/v4/objects/job_token_scope.py +++ b/gitlab/v4/objects/job_token_scope.py @@ -1,3 +1,5 @@ +from __future__ import annotations + from typing import cast from gitlab.base import RESTObject @@ -23,8 +25,8 @@ class ProjectJobTokenScope(RefreshMixin, SaveMixin, RESTObject): _id_attr = None - allowlist: "AllowlistProjectManager" - groups_allowlist: "AllowlistGroupManager" + allowlist: AllowlistProjectManager + groups_allowlist: AllowlistGroupManager class ProjectJobTokenScopeManager( diff --git a/gitlab/v4/objects/jobs.py b/gitlab/v4/objects/jobs.py index 69d83aae1..ff21581e9 100644 --- a/gitlab/v4/objects/jobs.py +++ b/gitlab/v4/objects/jobs.py @@ -1,13 +1,12 @@ +from __future__ import annotations + from typing import ( Any, Callable, - Dict, Iterator, Literal, - Optional, overload, TYPE_CHECKING, - Union, ) import requests @@ -28,7 +27,7 @@ class ProjectJob(RefreshMixin, RESTObject): @cli.register_custom_action(cls_names="ProjectJob") @exc.on_http_error(exc.GitlabJobCancelError) - def cancel(self, **kwargs: Any) -> Dict[str, Any]: + def cancel(self, **kwargs: Any) -> dict[str, Any]: """Cancel the job. Args: @@ -46,7 +45,7 @@ def cancel(self, **kwargs: Any) -> Dict[str, Any]: @cli.register_custom_action(cls_names="ProjectJob") @exc.on_http_error(exc.GitlabJobRetryError) - def retry(self, **kwargs: Any) -> Dict[str, Any]: + def retry(self, **kwargs: Any) -> dict[str, Any]: """Retry the job. Args: @@ -151,7 +150,7 @@ def artifacts( def artifacts( self, streamed: Literal[True] = True, - action: Optional[Callable[[bytes], Any]] = None, + action: Callable[[bytes], Any] | None = None, chunk_size: int = 1024, *, iterator: Literal[False] = False, @@ -163,12 +162,12 @@ def artifacts( def artifacts( self, streamed: bool = False, - action: Optional[Callable[..., Any]] = None, + action: Callable[..., Any] | None = None, chunk_size: int = 1024, *, iterator: bool = False, **kwargs: Any, - ) -> Optional[Union[bytes, Iterator[Any]]]: + ) -> bytes | Iterator[Any] | None: """Get the job artifacts. Args: @@ -228,7 +227,7 @@ def artifact( self, path: str, streamed: Literal[True] = True, - action: Optional[Callable[[bytes], Any]] = None, + action: Callable[[bytes], Any] | None = None, chunk_size: int = 1024, *, iterator: Literal[False] = False, @@ -241,12 +240,12 @@ def artifact( self, path: str, streamed: bool = False, - action: Optional[Callable[..., Any]] = None, + action: Callable[..., Any] | None = None, chunk_size: int = 1024, *, iterator: bool = False, **kwargs: Any, - ) -> Optional[Union[bytes, Iterator[Any]]]: + ) -> bytes | Iterator[Any] | None: """Get a single artifact file from within the job's artifacts archive. Args: @@ -304,7 +303,7 @@ def trace( def trace( self, streamed: Literal[True] = True, - action: Optional[Callable[[bytes], Any]] = None, + action: Callable[[bytes], Any] | None = None, chunk_size: int = 1024, *, iterator: Literal[False] = False, @@ -316,12 +315,12 @@ def trace( def trace( self, streamed: bool = False, - action: Optional[Callable[..., Any]] = None, + action: Callable[..., Any] | None = None, chunk_size: int = 1024, *, iterator: bool = False, **kwargs: Any, - ) -> Optional[Union[bytes, Iterator[Any]]]: + ) -> bytes | Iterator[Any] | None: """Get the job trace. Args: diff --git a/gitlab/v4/objects/keys.py b/gitlab/v4/objects/keys.py index 1bbc34095..46f9e9f08 100644 --- a/gitlab/v4/objects/keys.py +++ b/gitlab/v4/objects/keys.py @@ -1,4 +1,6 @@ -from typing import Any, Optional, TYPE_CHECKING, Union +from __future__ import annotations + +from typing import Any, TYPE_CHECKING from gitlab.base import RESTObject from gitlab.mixins import GetMixin @@ -18,7 +20,7 @@ class KeyManager(GetMixin[Key]): _obj_cls = Key def get( - self, id: Optional[Union[int, str]] = None, lazy: bool = False, **kwargs: Any + self, id: int | str | None = None, lazy: bool = False, **kwargs: Any ) -> Key: if id is not None: return super().get(id, lazy=lazy, **kwargs) diff --git a/gitlab/v4/objects/labels.py b/gitlab/v4/objects/labels.py index 488e2d05a..1d5db33f3 100644 --- a/gitlab/v4/objects/labels.py +++ b/gitlab/v4/objects/labels.py @@ -1,4 +1,6 @@ -from typing import Any, Dict, Optional +from __future__ import annotations + +from typing import Any from gitlab import exceptions as exc from gitlab.base import RESTObject @@ -24,7 +26,7 @@ class GroupLabel(SubscribableMixin, SaveMixin, ObjectDeleteMixin, RESTObject): _id_attr = "name" - manager: "GroupLabelManager" + manager: GroupLabelManager # Update without ID, but we need an ID to get from list. @exc.on_http_error(exc.GitlabUpdateError) @@ -67,11 +69,8 @@ class GroupLabelManager( # NOTE(jlvillal): Signature doesn't match UpdateMixin.update() so ignore # type error def update( # type: ignore[override] - self, - name: Optional[str], - new_data: Optional[Dict[str, Any]] = None, - **kwargs: Any, - ) -> Dict[str, Any]: + self, name: str | None, new_data: dict[str, Any] | None = None, **kwargs: Any + ) -> dict[str, Any]: """Update a Label on the server. Args: @@ -88,7 +87,7 @@ class ProjectLabel( PromoteMixin, SubscribableMixin, SaveMixin, ObjectDeleteMixin, RESTObject ): _id_attr = "name" - manager: "ProjectLabelManager" + manager: ProjectLabelManager # Update without ID, but we need an ID to get from list. @exc.on_http_error(exc.GitlabUpdateError) @@ -131,11 +130,8 @@ class ProjectLabelManager( # NOTE(jlvillal): Signature doesn't match UpdateMixin.update() so ignore # type error def update( # type: ignore[override] - self, - name: Optional[str], - new_data: Optional[Dict[str, Any]] = None, - **kwargs: Any, - ) -> Dict[str, Any]: + self, name: str | None, new_data: dict[str, Any] | None = None, **kwargs: Any + ) -> dict[str, Any]: """Update a Label on the server. Args: diff --git a/gitlab/v4/objects/ldap.py b/gitlab/v4/objects/ldap.py index 4f16e7010..584379fb3 100644 --- a/gitlab/v4/objects/ldap.py +++ b/gitlab/v4/objects/ldap.py @@ -1,4 +1,6 @@ -from typing import Any, List, Union +from __future__ import annotations + +from typing import Any from gitlab import exceptions as exc from gitlab.base import RESTManager, RESTObject, RESTObjectList @@ -19,7 +21,7 @@ class LDAPGroupManager(RESTManager[LDAPGroup]): _list_filters = ("search", "provider") @exc.on_http_error(exc.GitlabListError) - def list(self, **kwargs: Any) -> Union[List[LDAPGroup], RESTObjectList]: + def list(self, **kwargs: Any) -> list[LDAPGroup] | RESTObjectList: """Retrieve a list of objects. Args: diff --git a/gitlab/v4/objects/members.py b/gitlab/v4/objects/members.py index 9551e33a7..918e3c4ed 100644 --- a/gitlab/v4/objects/members.py +++ b/gitlab/v4/objects/members.py @@ -1,3 +1,5 @@ +from __future__ import annotations + from gitlab import types from gitlab.base import RESTObject from gitlab.mixins import ( @@ -51,7 +53,7 @@ class GroupMemberManager(CRUDMixin[GroupMember]): class GroupBillableMember(ObjectDeleteMixin, RESTObject): _repr_attr = "username" - memberships: "GroupBillableMemberMembershipManager" + memberships: GroupBillableMemberMembershipManager class GroupBillableMemberManager( diff --git a/gitlab/v4/objects/merge_request_approvals.py b/gitlab/v4/objects/merge_request_approvals.py index d160eae6f..c21421915 100644 --- a/gitlab/v4/objects/merge_request_approvals.py +++ b/gitlab/v4/objects/merge_request_approvals.py @@ -1,4 +1,6 @@ -from typing import Any, List, Optional, TYPE_CHECKING +from __future__ import annotations + +from typing import Any, TYPE_CHECKING from gitlab import exceptions as exc from gitlab.base import RESTObject @@ -110,11 +112,11 @@ class ProjectMergeRequestApprovalManager( def set_approvers( self, approvals_required: int, - approver_ids: Optional[List[int]] = None, - approver_group_ids: Optional[List[int]] = None, + approver_ids: list[int] | None = None, + approver_group_ids: list[int] | None = None, approval_rule_name: str = "name", *, - approver_usernames: Optional[List[str]] = None, + approver_usernames: list[str] | None = None, **kwargs: Any, ) -> RESTObject: """Change MR-level allowed approvers and approver groups. diff --git a/gitlab/v4/objects/merge_requests.py b/gitlab/v4/objects/merge_requests.py index 0ff9efae8..80a449143 100644 --- a/gitlab/v4/objects/merge_requests.py +++ b/gitlab/v4/objects/merge_requests.py @@ -4,7 +4,9 @@ https://docs.gitlab.com/ee/api/merge_request_approvals.html """ -from typing import Any, Dict, Optional, TYPE_CHECKING, Union +from __future__ import annotations + +from typing import Any, TYPE_CHECKING import requests @@ -159,7 +161,7 @@ class ProjectMergeRequest( approval_state: ProjectMergeRequestApprovalStateManager approvals: ProjectMergeRequestApprovalManager awardemojis: ProjectMergeRequestAwardEmojiManager - diffs: "ProjectMergeRequestDiffManager" + diffs: ProjectMergeRequestDiffManager discussions: ProjectMergeRequestDiscussionManager draft_notes: ProjectMergeRequestDraftNoteManager notes: ProjectMergeRequestNoteManager @@ -172,7 +174,7 @@ class ProjectMergeRequest( @cli.register_custom_action(cls_names="ProjectMergeRequest") @exc.on_http_error(exc.GitlabMROnBuildSuccessError) - def cancel_merge_when_pipeline_succeeds(self, **kwargs: Any) -> Dict[str, str]: + def cancel_merge_when_pipeline_succeeds(self, **kwargs: Any) -> dict[str, str]: """Cancel merge when the pipeline succeeds. Args: @@ -283,7 +285,7 @@ def commits(self, **kwargs: Any) -> RESTObjectList: cls_names="ProjectMergeRequest", optional=("access_raw_diffs",) ) @exc.on_http_error(exc.GitlabListError) - def changes(self, **kwargs: Any) -> Union[Dict[str, Any], requests.Response]: + def changes(self, **kwargs: Any) -> dict[str, Any] | requests.Response: """List the merge request changes. Args: @@ -301,7 +303,7 @@ def changes(self, **kwargs: Any) -> Union[Dict[str, Any], requests.Response]: @cli.register_custom_action(cls_names="ProjectMergeRequest", optional=("sha",)) @exc.on_http_error(exc.GitlabMRApprovalError) - def approve(self, sha: Optional[str] = None, **kwargs: Any) -> Dict[str, Any]: + def approve(self, sha: str | None = None, **kwargs: Any) -> dict[str, Any]: """Approve the merge request. Args: @@ -343,7 +345,7 @@ def unapprove(self, **kwargs: Any) -> None: https://docs.gitlab.com/ee/api/merge_request_approvals.html#unapprove-merge-request """ path = f"{self.manager.path}/{self.encoded_id}/unapprove" - data: Dict[str, Any] = {} + data: dict[str, Any] = {} server_data = self.manager.gitlab.http_post(path, post_data=data, **kwargs) if TYPE_CHECKING: @@ -352,7 +354,7 @@ def unapprove(self, **kwargs: Any) -> None: @cli.register_custom_action(cls_names="ProjectMergeRequest") @exc.on_http_error(exc.GitlabMRRebaseError) - def rebase(self, **kwargs: Any) -> Union[Dict[str, Any], requests.Response]: + def rebase(self, **kwargs: Any) -> dict[str, Any] | requests.Response: """Attempt to rebase the source branch onto the target branch Args: @@ -363,14 +365,12 @@ def rebase(self, **kwargs: Any) -> Union[Dict[str, Any], requests.Response]: GitlabMRRebaseError: If rebasing failed """ path = f"{self.manager.path}/{self.encoded_id}/rebase" - data: Dict[str, Any] = {} + data: dict[str, Any] = {} return self.manager.gitlab.http_put(path, post_data=data, **kwargs) @cli.register_custom_action(cls_names="ProjectMergeRequest") @exc.on_http_error(exc.GitlabMRResetApprovalError) - def reset_approvals( - self, **kwargs: Any - ) -> Union[Dict[str, Any], requests.Response]: + def reset_approvals(self, **kwargs: Any) -> dict[str, Any] | requests.Response: """Clear all approvals of the merge request. Args: @@ -381,12 +381,12 @@ def reset_approvals( GitlabMRResetApprovalError: If reset approval failed """ path = f"{self.manager.path}/{self.encoded_id}/reset_approvals" - data: Dict[str, Any] = {} + data: dict[str, Any] = {} return self.manager.gitlab.http_put(path, post_data=data, **kwargs) @cli.register_custom_action(cls_names="ProjectMergeRequest") @exc.on_http_error(exc.GitlabGetError) - def merge_ref(self, **kwargs: Any) -> Union[Dict[str, Any], requests.Response]: + def merge_ref(self, **kwargs: Any) -> dict[str, Any] | requests.Response: """Attempt to merge changes between source and target branches into `refs/merge-requests/:iid/merge`. @@ -410,11 +410,11 @@ def merge_ref(self, **kwargs: Any) -> Union[Dict[str, Any], requests.Response]: @exc.on_http_error(exc.GitlabMRClosedError) def merge( self, - merge_commit_message: Optional[str] = None, - should_remove_source_branch: Optional[bool] = None, - merge_when_pipeline_succeeds: Optional[bool] = None, + merge_commit_message: str | None = None, + should_remove_source_branch: bool | None = None, + merge_when_pipeline_succeeds: bool | None = None, **kwargs: Any, - ) -> Dict[str, Any]: + ) -> dict[str, Any]: """Accept the merge request. Args: @@ -430,7 +430,7 @@ def merge( GitlabMRClosedError: If the merge failed """ path = f"{self.manager.path}/{self.encoded_id}/merge" - data: Dict[str, Any] = {} + data: dict[str, Any] = {} if merge_commit_message: data["merge_commit_message"] = merge_commit_message if should_remove_source_branch is not None: diff --git a/gitlab/v4/objects/packages.py b/gitlab/v4/objects/packages.py index fb2d9f797..773c9d0eb 100644 --- a/gitlab/v4/objects/packages.py +++ b/gitlab/v4/objects/packages.py @@ -4,6 +4,8 @@ https://docs.gitlab.com/ee/user/packages/generic_packages/ """ +from __future__ import annotations + from pathlib import Path from typing import ( Any, @@ -11,10 +13,8 @@ Callable, Iterator, Literal, - Optional, overload, TYPE_CHECKING, - Union, ) import requests @@ -58,9 +58,9 @@ def upload( package_name: str, package_version: str, file_name: str, - path: Optional[Union[str, Path]] = None, - select: Optional[str] = None, - data: Optional[Union[bytes, BinaryIO]] = None, + path: str | Path | None = None, + select: str | None = None, + data: bytes | BinaryIO | None = None, **kwargs: Any, ) -> GenericPackage: """Upload a file as a generic package. @@ -92,7 +92,7 @@ def upload( if path is not None and data is not None: raise exc.GitlabUploadError("File contents and file path specified") - file_data: Optional[Union[bytes, BinaryIO]] = data + file_data: bytes | BinaryIO | None = data if not file_data: if TYPE_CHECKING: @@ -158,7 +158,7 @@ def download( package_version: str, file_name: str, streamed: Literal[True] = True, - action: Optional[Callable[[bytes], Any]] = None, + action: Callable[[bytes], Any] | None = None, chunk_size: int = 1024, *, iterator: Literal[False] = False, @@ -176,12 +176,12 @@ def download( package_version: str, file_name: str, streamed: bool = False, - action: Optional[Callable[[bytes], Any]] = None, + action: Callable[[bytes], Any] | None = None, chunk_size: int = 1024, *, iterator: bool = False, **kwargs: Any, - ) -> Optional[Union[bytes, Iterator[Any]]]: + ) -> bytes | Iterator[Any] | None: """Download a generic package. Args: @@ -232,8 +232,8 @@ class GroupPackageManager(ListMixin[GroupPackage]): class ProjectPackage(ObjectDeleteMixin, RESTObject): - package_files: "ProjectPackageFileManager" - pipelines: "ProjectPackagePipelineManager" + package_files: ProjectPackageFileManager + pipelines: ProjectPackagePipelineManager class ProjectPackageManager( diff --git a/gitlab/v4/objects/pipelines.py b/gitlab/v4/objects/pipelines.py index 2645b7064..14030c167 100644 --- a/gitlab/v4/objects/pipelines.py +++ b/gitlab/v4/objects/pipelines.py @@ -1,4 +1,6 @@ -from typing import Any, Dict, Optional, TYPE_CHECKING, Union +from __future__ import annotations + +from typing import Any, TYPE_CHECKING import requests @@ -56,15 +58,15 @@ class ProjectMergeRequestPipelineManager( class ProjectPipeline(RefreshMixin, ObjectDeleteMixin, RESTObject): - bridges: "ProjectPipelineBridgeManager" - jobs: "ProjectPipelineJobManager" - test_report: "ProjectPipelineTestReportManager" - test_report_summary: "ProjectPipelineTestReportSummaryManager" - variables: "ProjectPipelineVariableManager" + bridges: ProjectPipelineBridgeManager + jobs: ProjectPipelineJobManager + test_report: ProjectPipelineTestReportManager + test_report_summary: ProjectPipelineTestReportSummaryManager + variables: ProjectPipelineVariableManager @cli.register_custom_action(cls_names="ProjectPipeline") @exc.on_http_error(exc.GitlabPipelineCancelError) - def cancel(self, **kwargs: Any) -> Union[Dict[str, Any], requests.Response]: + def cancel(self, **kwargs: Any) -> dict[str, Any] | requests.Response: """Cancel the job. Args: @@ -79,7 +81,7 @@ def cancel(self, **kwargs: Any) -> Union[Dict[str, Any], requests.Response]: @cli.register_custom_action(cls_names="ProjectPipeline") @exc.on_http_error(exc.GitlabPipelineRetryError) - def retry(self, **kwargs: Any) -> Union[Dict[str, Any], requests.Response]: + def retry(self, **kwargs: Any) -> dict[str, Any] | requests.Response: """Retry the job. Args: @@ -116,7 +118,7 @@ class ProjectPipelineManager( _create_attrs = RequiredOptional(required=("ref",)) def create( - self, data: Optional[Dict[str, Any]] = None, **kwargs: Any + self, data: dict[str, Any] | None = None, **kwargs: Any ) -> ProjectPipeline: """Creates a new object. @@ -136,7 +138,7 @@ def create( path = self.path[:-1] # drop the 's' return super().create(data, path=path, **kwargs) - def latest(self, ref: Optional[str] = None, lazy: bool = False) -> ProjectPipeline: + def latest(self, ref: str | None = None, lazy: bool = False) -> ProjectPipeline: """Get the latest pipeline for the most recent commit on a specific ref in a project @@ -240,7 +242,7 @@ def take_ownership(self, **kwargs: Any) -> None: @cli.register_custom_action(cls_names="ProjectPipelineSchedule") @exc.on_http_error(exc.GitlabPipelinePlayError) - def play(self, **kwargs: Any) -> Dict[str, Any]: + def play(self, **kwargs: Any) -> dict[str, Any]: """Trigger a new scheduled pipeline, which runs immediately. The next scheduled run of this pipeline is not affected. diff --git a/gitlab/v4/objects/projects.py b/gitlab/v4/objects/projects.py index ea211cba5..76a903b31 100644 --- a/gitlab/v4/objects/projects.py +++ b/gitlab/v4/objects/projects.py @@ -3,18 +3,16 @@ https://docs.gitlab.com/ee/api/projects.html """ +from __future__ import annotations + import io from typing import ( Any, Callable, - Dict, Iterator, - List, Literal, - Optional, overload, TYPE_CHECKING, - Union, ) import requests @@ -209,7 +207,7 @@ class Project( events: ProjectEventManager exports: ProjectExportManager files: ProjectFileManager - forks: "ProjectForkManager" + forks: ProjectForkManager generic_packages: GenericPackageManager gitignore_templates: ProjectGitignoreTemplateManager gitlabciyml_templates: ProjectGitlabciymlTemplateManager @@ -249,15 +247,15 @@ class Project( registry_protection_repository_rules: ProjectRegistryRepositoryProtectionRuleManager releases: ProjectReleaseManager resource_groups: ProjectResourceGroupManager - remote_mirrors: "ProjectRemoteMirrorManager" - pull_mirror: "ProjectPullMirrorManager" + remote_mirrors: ProjectRemoteMirrorManager + pull_mirror: ProjectPullMirrorManager repositories: ProjectRegistryRepositoryManager runners: ProjectRunnerManager secure_files: ProjectSecureFileManager services: ProjectServiceManager snippets: ProjectSnippetManager external_status_checks: ProjectExternalStatusCheckManager - storage: "ProjectStorageManager" + storage: ProjectStorageManager tags: ProjectTagManager triggers: ProjectTriggerManager users: ProjectUserManager @@ -297,7 +295,7 @@ def delete_fork_relation(self, **kwargs: Any) -> None: @cli.register_custom_action(cls_names="Project") @exc.on_http_error(exc.GitlabGetError) - def languages(self, **kwargs: Any) -> Union[Dict[str, Any], requests.Response]: + def languages(self, **kwargs: Any) -> dict[str, Any] | requests.Response: """Get languages used in the project with percentage value. Args: @@ -392,7 +390,7 @@ def share( self, group_id: int, group_access: int, - expires_at: Optional[str] = None, + expires_at: str | None = None, **kwargs: Any, ) -> None: """Share the project with a group. @@ -437,7 +435,7 @@ def trigger_pipeline( self, ref: str, token: str, - variables: Optional[Dict[str, Any]] = None, + variables: dict[str, Any] | None = None, **kwargs: Any, ) -> ProjectPipeline: """Trigger a CI build. @@ -522,7 +520,7 @@ def snapshot( self, wiki: bool = False, streamed: Literal[True] = True, - action: Optional[Callable[[bytes], Any]] = None, + action: Callable[[bytes], Any] | None = None, chunk_size: int = 1024, *, iterator: Literal[False] = False, @@ -535,12 +533,12 @@ def snapshot( self, wiki: bool = False, streamed: bool = False, - action: Optional[Callable[[bytes], Any]] = None, + action: Callable[[bytes], Any] | None = None, chunk_size: int = 1024, *, iterator: bool = False, **kwargs: Any, - ) -> Optional[Union[bytes, Iterator[Any]]]: + ) -> bytes | Iterator[Any] | None: """Return a snapshot of the repository. Args: @@ -576,7 +574,7 @@ def snapshot( @exc.on_http_error(exc.GitlabSearchError) def search( self, scope: str, search: str, **kwargs: Any - ) -> Union[client.GitlabList, List[Dict[str, Any]]]: + ) -> client.GitlabList | list[dict[str, Any]]: """Search the project resources matching the provided string.' Args: @@ -619,7 +617,7 @@ def mirror_pull(self, **kwargs: Any) -> None: @cli.register_custom_action(cls_names="Project") @exc.on_http_error(exc.GitlabGetError) - def mirror_pull_details(self, **kwargs: Any) -> Dict[str, Any]: + def mirror_pull_details(self, **kwargs: Any) -> dict[str, Any]: """Get a project's pull mirror details. Introduced in GitLab 15.5. @@ -649,7 +647,7 @@ def mirror_pull_details(self, **kwargs: Any) -> Dict[str, Any]: @cli.register_custom_action(cls_names="Project", required=("to_namespace",)) @exc.on_http_error(exc.GitlabTransferProjectError) - def transfer(self, to_namespace: Union[int, str], **kwargs: Any) -> None: + def transfer(self, to_namespace: int | str, **kwargs: Any) -> None: """Transfer a project to the given namespace ID Args: @@ -871,12 +869,12 @@ def import_project( self, file: io.BufferedReader, path: str, - name: Optional[str] = None, - namespace: Optional[str] = None, + name: str | None = None, + namespace: str | None = None, overwrite: bool = False, - override_params: Optional[Dict[str, Any]] = None, + override_params: dict[str, Any] | None = None, **kwargs: Any, - ) -> Union[Dict[str, Any], requests.Response]: + ) -> dict[str, Any] | requests.Response: """Import a project from an archive file. Args: @@ -916,12 +914,12 @@ def remote_import( self, url: str, path: str, - name: Optional[str] = None, - namespace: Optional[str] = None, + name: str | None = None, + namespace: str | None = None, overwrite: bool = False, - override_params: Optional[Dict[str, Any]] = None, + override_params: dict[str, Any] | None = None, **kwargs: Any, - ) -> Union[Dict[str, Any], requests.Response]: + ) -> dict[str, Any] | requests.Response: """Import a project from an archive file stored on a remote URL. Args: @@ -964,12 +962,12 @@ def remote_import_s3( file_key: str, access_key_id: str, secret_access_key: str, - name: Optional[str] = None, - namespace: Optional[str] = None, + name: str | None = None, + namespace: str | None = None, overwrite: bool = False, - override_params: Optional[Dict[str, Any]] = None, + override_params: dict[str, Any] | None = None, **kwargs: Any, - ) -> Union[Dict[str, Any], requests.Response]: + ) -> dict[str, Any] | requests.Response: """Import a project from an archive file stored on AWS S3. Args: @@ -1022,10 +1020,10 @@ def import_bitbucket_server( personal_access_token: str, bitbucket_server_project: str, bitbucket_server_repo: str, - new_name: Optional[str] = None, - target_namespace: Optional[str] = None, + new_name: str | None = None, + target_namespace: str | None = None, **kwargs: Any, - ) -> Union[Dict[str, Any], requests.Response]: + ) -> dict[str, Any] | requests.Response: """Import a project from BitBucket Server to Gitlab (schedule the import) This method will return when an import operation has been safely queued, @@ -1112,11 +1110,11 @@ def import_github( personal_access_token: str, repo_id: int, target_namespace: str, - new_name: Optional[str] = None, - github_hostname: Optional[str] = None, - optional_stages: Optional[Dict[str, bool]] = None, + new_name: str | None = None, + github_hostname: str | None = None, + optional_stages: dict[str, bool] | None = None, **kwargs: Any, - ) -> Union[Dict[str, Any], requests.Response]: + ) -> dict[str, Any] | requests.Response: """Import a project from Github to Gitlab (schedule the import) This method will return when an import operation has been safely queued, @@ -1213,9 +1211,7 @@ class ProjectForkManager(CreateMixin[ProjectFork], ListMixin[ProjectFork]): ) _create_attrs = RequiredOptional(optional=("namespace",)) - def create( - self, data: Optional[Dict[str, Any]] = None, **kwargs: Any - ) -> ProjectFork: + def create(self, data: dict[str, Any] | None = None, **kwargs: Any) -> ProjectFork: """Creates a new object. Args: @@ -1267,7 +1263,7 @@ class ProjectPullMirrorManager( _update_attrs = RequiredOptional(optional=("url",)) @exc.on_http_error(exc.GitlabCreateError) - def create(self, data: Dict[str, Any], **kwargs: Any) -> ProjectPullMirror: + def create(self, data: dict[str, Any], **kwargs: Any) -> ProjectPullMirror: """Create a new object. Args: diff --git a/gitlab/v4/objects/releases.py b/gitlab/v4/objects/releases.py index 30794e2aa..e1976c568 100644 --- a/gitlab/v4/objects/releases.py +++ b/gitlab/v4/objects/releases.py @@ -1,3 +1,5 @@ +from __future__ import annotations + from gitlab.base import RESTObject from gitlab.mixins import CRUDMixin, ObjectDeleteMixin, SaveMixin from gitlab.types import ArrayAttribute, RequiredOptional @@ -13,7 +15,7 @@ class ProjectRelease(SaveMixin, RESTObject): _id_attr = "tag_name" - links: "ProjectReleaseLinkManager" + links: ProjectReleaseLinkManager class ProjectReleaseManager(CRUDMixin[ProjectRelease]): diff --git a/gitlab/v4/objects/repositories.py b/gitlab/v4/objects/repositories.py index aece75d74..1d2c1e964 100644 --- a/gitlab/v4/objects/repositories.py +++ b/gitlab/v4/objects/repositories.py @@ -4,17 +4,15 @@ Currently this module only contains repository-related methods for projects. """ +from __future__ import annotations + from typing import ( Any, Callable, - Dict, Iterator, - List, Literal, - Optional, overload, TYPE_CHECKING, - Union, ) import requests @@ -38,7 +36,7 @@ class RepositoryMixin(_RestObjectBase): @exc.on_http_error(exc.GitlabUpdateError) def update_submodule( self, submodule: str, branch: str, commit_sha: str, **kwargs: Any - ) -> Union[Dict[str, Any], requests.Response]: + ) -> dict[str, Any] | requests.Response: """Update a project submodule Args: @@ -66,7 +64,7 @@ def update_submodule( @exc.on_http_error(exc.GitlabGetError) def repository_tree( self, path: str = "", ref: str = "", recursive: bool = False, **kwargs: Any - ) -> Union[gitlab.client.GitlabList, List[Dict[str, Any]]]: + ) -> gitlab.client.GitlabList | list[dict[str, Any]]: """Return a list of files in the repository. Args: @@ -88,7 +86,7 @@ def repository_tree( The representation of the tree """ gl_path = f"/projects/{self.encoded_id}/repository/tree" - query_data: Dict[str, Any] = {"recursive": recursive} + query_data: dict[str, Any] = {"recursive": recursive} if path: query_data["path"] = path if ref: @@ -99,7 +97,7 @@ def repository_tree( @exc.on_http_error(exc.GitlabGetError) def repository_blob( self, sha: str, **kwargs: Any - ) -> Union[Dict[str, Any], requests.Response]: + ) -> dict[str, Any] | requests.Response: """Return a file by blob SHA. Args: @@ -146,7 +144,7 @@ def repository_raw_blob( self, sha: str, streamed: Literal[True] = True, - action: Optional[Callable[[bytes], Any]] = None, + action: Callable[[bytes], Any] | None = None, chunk_size: int = 1024, *, iterator: Literal[False] = False, @@ -159,12 +157,12 @@ def repository_raw_blob( self, sha: str, streamed: bool = False, - action: Optional[Callable[..., Any]] = None, + action: Callable[..., Any] | None = None, chunk_size: int = 1024, *, iterator: bool = False, **kwargs: Any, - ) -> Optional[Union[bytes, Iterator[Any]]]: + ) -> bytes | Iterator[Any] | None: """Return the raw file contents for a blob. Args: @@ -200,7 +198,7 @@ def repository_raw_blob( @exc.on_http_error(exc.GitlabGetError) def repository_compare( self, from_: str, to: str, **kwargs: Any - ) -> Union[Dict[str, Any], requests.Response]: + ) -> dict[str, Any] | requests.Response: """Return a diff between two branches/commits. Args: @@ -223,7 +221,7 @@ def repository_compare( @exc.on_http_error(exc.GitlabGetError) def repository_contributors( self, **kwargs: Any - ) -> Union[gitlab.client.GitlabList, List[Dict[str, Any]]]: + ) -> gitlab.client.GitlabList | list[dict[str, Any]]: """Return a list of contributors for the project. Args: @@ -247,7 +245,7 @@ def repository_contributors( @overload def repository_archive( self, - sha: Optional[str] = None, + sha: str | None = None, streamed: Literal[False] = False, action: None = None, chunk_size: int = 1024, @@ -259,7 +257,7 @@ def repository_archive( @overload def repository_archive( self, - sha: Optional[str] = None, + sha: str | None = None, streamed: bool = False, action: None = None, chunk_size: int = 1024, @@ -271,9 +269,9 @@ def repository_archive( @overload def repository_archive( self, - sha: Optional[str] = None, + sha: str | None = None, streamed: Literal[True] = True, - action: Optional[Callable[[bytes], Any]] = None, + action: Callable[[bytes], Any] | None = None, chunk_size: int = 1024, *, iterator: Literal[False] = False, @@ -284,16 +282,16 @@ def repository_archive( @exc.on_http_error(exc.GitlabListError) def repository_archive( self, - sha: Optional[str] = None, + sha: str | None = None, streamed: bool = False, - action: Optional[Callable[..., Any]] = None, + action: Callable[..., Any] | None = None, chunk_size: int = 1024, - format: Optional[str] = None, - path: Optional[str] = None, + format: str | None = None, + path: str | None = None, *, iterator: bool = False, **kwargs: Any, - ) -> Optional[Union[bytes, Iterator[Any]]]: + ) -> bytes | Iterator[Any] | None: """Return an archive of the repository. Args: @@ -337,8 +335,8 @@ def repository_archive( @cli.register_custom_action(cls_names="Project", required=("refs",)) @exc.on_http_error(exc.GitlabGetError) def repository_merge_base( - self, refs: List[str], **kwargs: Any - ) -> Union[Dict[str, Any], requests.Response]: + self, refs: list[str], **kwargs: Any + ) -> dict[str, Any] | requests.Response: """Return a diff between two branches/commits. Args: diff --git a/gitlab/v4/objects/resource_groups.py b/gitlab/v4/objects/resource_groups.py index ef8e444c9..82d3a745d 100644 --- a/gitlab/v4/objects/resource_groups.py +++ b/gitlab/v4/objects/resource_groups.py @@ -1,3 +1,5 @@ +from __future__ import annotations + from gitlab.base import RESTObject from gitlab.mixins import ListMixin, RetrieveMixin, SaveMixin, UpdateMixin from gitlab.types import RequiredOptional @@ -13,7 +15,7 @@ class ProjectResourceGroup(SaveMixin, RESTObject): _id_attr = "key" - upcoming_jobs: "ProjectResourceGroupUpcomingJobManager" + upcoming_jobs: ProjectResourceGroupUpcomingJobManager class ProjectResourceGroupManager( diff --git a/gitlab/v4/objects/runners.py b/gitlab/v4/objects/runners.py index faa82ddbd..4d6040a82 100644 --- a/gitlab/v4/objects/runners.py +++ b/gitlab/v4/objects/runners.py @@ -1,4 +1,6 @@ -from typing import Any, List, Optional +from __future__ import annotations + +from typing import Any from gitlab import cli from gitlab import exceptions as exc @@ -76,7 +78,7 @@ class RunnerManager(CRUDMixin[Runner]): @cli.register_custom_action(cls_names="RunnerManager", optional=("scope",)) @exc.on_http_error(exc.GitlabListError) - def all(self, scope: Optional[str] = None, **kwargs: Any) -> List[Runner]: + def all(self, scope: str | None = None, **kwargs: Any) -> list[Runner]: """List all the runners. Args: diff --git a/gitlab/v4/objects/secure_files.py b/gitlab/v4/objects/secure_files.py index 6ced2c306..78e516a33 100644 --- a/gitlab/v4/objects/secure_files.py +++ b/gitlab/v4/objects/secure_files.py @@ -3,15 +3,15 @@ https://docs.gitlab.com/ee/api/secure_files.html """ +from __future__ import annotations + from typing import ( Any, Callable, Iterator, Literal, - Optional, overload, TYPE_CHECKING, - Union, ) import requests @@ -53,7 +53,7 @@ def download( def download( self, streamed: Literal[True] = True, - action: Optional[Callable[[bytes], Any]] = None, + action: Callable[[bytes], Any] | None = None, chunk_size: int = 1024, *, iterator: Literal[False] = False, @@ -65,12 +65,12 @@ def download( def download( self, streamed: bool = False, - action: Optional[Callable[[bytes], Any]] = None, + action: Callable[[bytes], Any] | None = None, chunk_size: int = 1024, *, iterator: bool = False, **kwargs: Any, - ) -> Optional[Union[bytes, Iterator[Any]]]: + ) -> bytes | Iterator[Any] | None: """Download the secure file. Args: diff --git a/gitlab/v4/objects/settings.py b/gitlab/v4/objects/settings.py index 63777420e..e415e6c79 100644 --- a/gitlab/v4/objects/settings.py +++ b/gitlab/v4/objects/settings.py @@ -1,4 +1,6 @@ -from typing import Any, Dict, Optional, Union +from __future__ import annotations + +from typing import Any from gitlab import exceptions as exc from gitlab import types @@ -94,10 +96,10 @@ class ApplicationSettingsManager( @exc.on_http_error(exc.GitlabUpdateError) def update( self, - id: Optional[Union[str, int]] = None, - new_data: Optional[Dict[str, Any]] = None, + id: str | int | None = None, + new_data: dict[str, Any] | None = None, **kwargs: Any, - ) -> Dict[str, Any]: + ) -> dict[str, Any]: """Update an object on the server. Args: diff --git a/gitlab/v4/objects/sidekiq.py b/gitlab/v4/objects/sidekiq.py index 8f89971f8..f99749a58 100644 --- a/gitlab/v4/objects/sidekiq.py +++ b/gitlab/v4/objects/sidekiq.py @@ -1,4 +1,6 @@ -from typing import Any, Dict, Union +from __future__ import annotations + +from typing import Any import requests @@ -23,7 +25,7 @@ class SidekiqManager(RESTManager[RESTObject]): @cli.register_custom_action(cls_names="SidekiqManager") @exc.on_http_error(exc.GitlabGetError) - def queue_metrics(self, **kwargs: Any) -> Union[Dict[str, Any], requests.Response]: + def queue_metrics(self, **kwargs: Any) -> dict[str, Any] | requests.Response: """Return the registered queues information. Args: @@ -40,9 +42,7 @@ def queue_metrics(self, **kwargs: Any) -> Union[Dict[str, Any], requests.Respons @cli.register_custom_action(cls_names="SidekiqManager") @exc.on_http_error(exc.GitlabGetError) - def process_metrics( - self, **kwargs: Any - ) -> Union[Dict[str, Any], requests.Response]: + def process_metrics(self, **kwargs: Any) -> dict[str, Any] | requests.Response: """Return the registered sidekiq workers. Args: @@ -59,7 +59,7 @@ def process_metrics( @cli.register_custom_action(cls_names="SidekiqManager") @exc.on_http_error(exc.GitlabGetError) - def job_stats(self, **kwargs: Any) -> Union[Dict[str, Any], requests.Response]: + def job_stats(self, **kwargs: Any) -> dict[str, Any] | requests.Response: """Return statistics about the jobs performed. Args: @@ -76,9 +76,7 @@ def job_stats(self, **kwargs: Any) -> Union[Dict[str, Any], requests.Response]: @cli.register_custom_action(cls_names="SidekiqManager") @exc.on_http_error(exc.GitlabGetError) - def compound_metrics( - self, **kwargs: Any - ) -> Union[Dict[str, Any], requests.Response]: + def compound_metrics(self, **kwargs: Any) -> dict[str, Any] | requests.Response: """Return all available metrics and statistics. Args: diff --git a/gitlab/v4/objects/snippets.py b/gitlab/v4/objects/snippets.py index 563f43f95..3e20accac 100644 --- a/gitlab/v4/objects/snippets.py +++ b/gitlab/v4/objects/snippets.py @@ -1,13 +1,12 @@ +from __future__ import annotations + from typing import ( Any, Callable, Iterator, - List, Literal, - Optional, overload, TYPE_CHECKING, - Union, ) import requests @@ -60,7 +59,7 @@ def content( def content( self, streamed: Literal[True] = True, - action: Optional[Callable[[bytes], Any]] = None, + action: Callable[[bytes], Any] | None = None, chunk_size: int = 1024, *, iterator: Literal[False] = False, @@ -72,12 +71,12 @@ def content( def content( self, streamed: bool = False, - action: Optional[Callable[..., Any]] = None, + action: Callable[..., Any] | None = None, chunk_size: int = 1024, *, iterator: bool = False, **kwargs: Any, - ) -> Optional[Union[bytes, Iterator[Any]]]: + ) -> bytes | Iterator[Any] | None: """Return the content of a snippet. Args: @@ -133,7 +132,7 @@ class SnippetManager(CRUDMixin[Snippet]): ) @cli.register_custom_action(cls_names="SnippetManager") - def list_public(self, **kwargs: Any) -> Union[RESTObjectList, List[Snippet]]: + def list_public(self, **kwargs: Any) -> RESTObjectList | list[Snippet]: """List all public snippets. Args: @@ -153,7 +152,7 @@ def list_public(self, **kwargs: Any) -> Union[RESTObjectList, List[Snippet]]: return self.list(path="/snippets/public", **kwargs) @cli.register_custom_action(cls_names="SnippetManager") - def list_all(self, **kwargs: Any) -> Union[RESTObjectList, List[Snippet]]: + def list_all(self, **kwargs: Any) -> RESTObjectList | list[Snippet]: """List all snippets. Args: @@ -172,7 +171,7 @@ def list_all(self, **kwargs: Any) -> Union[RESTObjectList, List[Snippet]]: """ return self.list(path="/snippets/all", **kwargs) - def public(self, **kwargs: Any) -> Union[RESTObjectList, List[Snippet]]: + def public(self, **kwargs: Any) -> RESTObjectList | list[Snippet]: """List all public snippets. Args: @@ -233,7 +232,7 @@ def content( def content( self, streamed: Literal[True] = True, - action: Optional[Callable[[bytes], Any]] = None, + action: Callable[[bytes], Any] | None = None, chunk_size: int = 1024, *, iterator: Literal[False] = False, @@ -245,12 +244,12 @@ def content( def content( self, streamed: bool = False, - action: Optional[Callable[..., Any]] = None, + action: Callable[..., Any] | None = None, chunk_size: int = 1024, *, iterator: bool = False, **kwargs: Any, - ) -> Optional[Union[bytes, Iterator[Any]]]: + ) -> bytes | Iterator[Any] | None: """Return the content of a snippet. Args: diff --git a/gitlab/v4/objects/topics.py b/gitlab/v4/objects/topics.py index 914111e3e..83c0b5475 100644 --- a/gitlab/v4/objects/topics.py +++ b/gitlab/v4/objects/topics.py @@ -1,4 +1,6 @@ -from typing import Any, Dict, TYPE_CHECKING, Union +from __future__ import annotations + +from typing import Any, TYPE_CHECKING from gitlab import cli from gitlab import exceptions as exc @@ -35,11 +37,8 @@ class TopicManager(CRUDMixin[Topic]): ) @exc.on_http_error(exc.GitlabMRClosedError) def merge( - self, - source_topic_id: Union[int, str], - target_topic_id: Union[int, str], - **kwargs: Any, - ) -> Dict[str, Any]: + self, source_topic_id: int | str, target_topic_id: int | str, **kwargs: Any + ) -> dict[str, Any]: """Merge two topics, assigning all projects to the target topic. Args: diff --git a/gitlab/v4/objects/users.py b/gitlab/v4/objects/users.py index 6e10d8fad..05f6413dc 100644 --- a/gitlab/v4/objects/users.py +++ b/gitlab/v4/objects/users.py @@ -4,7 +4,9 @@ https://docs.gitlab.com/ee/api/projects.html#list-projects-starred-by-a-user """ -from typing import Any, cast, Dict, List, Optional, Union +from __future__ import annotations + +from typing import Any, cast, Optional import requests @@ -169,23 +171,23 @@ class User(SaveMixin, ObjectDeleteMixin, RESTObject): _repr_attr = "username" customattributes: UserCustomAttributeManager - emails: "UserEmailManager" + emails: UserEmailManager events: UserEventManager - followers_users: "UserFollowersManager" - following_users: "UserFollowingManager" - gpgkeys: "UserGPGKeyManager" - identityproviders: "UserIdentityProviderManager" - impersonationtokens: "UserImpersonationTokenManager" - keys: "UserKeyManager" - memberships: "UserMembershipManager" + followers_users: UserFollowersManager + following_users: UserFollowingManager + gpgkeys: UserGPGKeyManager + identityproviders: UserIdentityProviderManager + impersonationtokens: UserImpersonationTokenManager + keys: UserKeyManager + memberships: UserMembershipManager personal_access_tokens: UserPersonalAccessTokenManager - projects: "UserProjectManager" - starred_projects: "StarredProjectManager" - status: "UserStatusManager" + projects: UserProjectManager + starred_projects: StarredProjectManager + status: UserStatusManager @cli.register_custom_action(cls_names="User") @exc.on_http_error(exc.GitlabBlockError) - def block(self, **kwargs: Any) -> Optional[bool]: + def block(self, **kwargs: Any) -> bool | None: """Block the user. Args: @@ -210,7 +212,7 @@ def block(self, **kwargs: Any) -> Optional[bool]: @cli.register_custom_action(cls_names="User") @exc.on_http_error(exc.GitlabFollowError) - def follow(self, **kwargs: Any) -> Union[Dict[str, Any], requests.Response]: + def follow(self, **kwargs: Any) -> dict[str, Any] | requests.Response: """Follow the user. Args: @@ -228,7 +230,7 @@ def follow(self, **kwargs: Any) -> Union[Dict[str, Any], requests.Response]: @cli.register_custom_action(cls_names="User") @exc.on_http_error(exc.GitlabUnfollowError) - def unfollow(self, **kwargs: Any) -> Union[Dict[str, Any], requests.Response]: + def unfollow(self, **kwargs: Any) -> dict[str, Any] | requests.Response: """Unfollow the user. Args: @@ -246,7 +248,7 @@ def unfollow(self, **kwargs: Any) -> Union[Dict[str, Any], requests.Response]: @cli.register_custom_action(cls_names="User") @exc.on_http_error(exc.GitlabUnblockError) - def unblock(self, **kwargs: Any) -> Optional[bool]: + def unblock(self, **kwargs: Any) -> bool | None: """Unblock the user. Args: @@ -271,7 +273,7 @@ def unblock(self, **kwargs: Any) -> Optional[bool]: @cli.register_custom_action(cls_names="User") @exc.on_http_error(exc.GitlabDeactivateError) - def deactivate(self, **kwargs: Any) -> Union[Dict[str, Any], requests.Response]: + def deactivate(self, **kwargs: Any) -> dict[str, Any] | requests.Response: """Deactivate the user. Args: @@ -292,7 +294,7 @@ def deactivate(self, **kwargs: Any) -> Union[Dict[str, Any], requests.Response]: @cli.register_custom_action(cls_names="User") @exc.on_http_error(exc.GitlabActivateError) - def activate(self, **kwargs: Any) -> Union[Dict[str, Any], requests.Response]: + def activate(self, **kwargs: Any) -> dict[str, Any] | requests.Response: """Activate the user. Args: @@ -313,7 +315,7 @@ def activate(self, **kwargs: Any) -> Union[Dict[str, Any], requests.Response]: @cli.register_custom_action(cls_names="User") @exc.on_http_error(exc.GitlabUserApproveError) - def approve(self, **kwargs: Any) -> Union[Dict[str, Any], requests.Response]: + def approve(self, **kwargs: Any) -> dict[str, Any] | requests.Response: """Approve a user creation request. Args: @@ -331,7 +333,7 @@ def approve(self, **kwargs: Any) -> Union[Dict[str, Any], requests.Response]: @cli.register_custom_action(cls_names="User") @exc.on_http_error(exc.GitlabUserRejectError) - def reject(self, **kwargs: Any) -> Union[Dict[str, Any], requests.Response]: + def reject(self, **kwargs: Any) -> dict[str, Any] | requests.Response: """Reject a user creation request. Args: @@ -349,7 +351,7 @@ def reject(self, **kwargs: Any) -> Union[Dict[str, Any], requests.Response]: @cli.register_custom_action(cls_names="User") @exc.on_http_error(exc.GitlabBanError) - def ban(self, **kwargs: Any) -> Union[Dict[str, Any], requests.Response]: + def ban(self, **kwargs: Any) -> dict[str, Any] | requests.Response: """Ban the user. Args: @@ -370,7 +372,7 @@ def ban(self, **kwargs: Any) -> Union[Dict[str, Any], requests.Response]: @cli.register_custom_action(cls_names="User") @exc.on_http_error(exc.GitlabUnbanError) - def unban(self, **kwargs: Any) -> Union[Dict[str, Any], requests.Response]: + def unban(self, **kwargs: Any) -> dict[str, Any] | requests.Response: """Unban the user. Args: @@ -621,7 +623,7 @@ class UserProjectManager(ListMixin[UserProject], CreateMixin[UserProject]): "id_before", ) - def list(self, **kwargs: Any) -> Union[RESTObjectList, List[UserProject]]: + def list(self, **kwargs: Any) -> RESTObjectList | list[UserProject]: """Retrieve a list of objects. Args: diff --git a/tests/functional/api/test_packages.py b/tests/functional/api/test_packages.py index eba35d316..535c2031f 100644 --- a/tests/functional/api/test_packages.py +++ b/tests/functional/api/test_packages.py @@ -116,7 +116,7 @@ def test_stream_generic_package(project): assert isinstance(bytes_iterator, Iterator) - package = bytes() + package = b"" for chunk in bytes_iterator: package += chunk @@ -136,7 +136,7 @@ def test_download_generic_package_to_file(tmp_path, project): action=f.write, ) - with open(path, "r") as f: + with open(path) as f: assert f.read() == file_content @@ -154,7 +154,7 @@ def test_stream_generic_package_to_file(tmp_path, project): for chunk in bytes_iterator: f.write(chunk) - with open(path, "r") as f: + with open(path) as f: assert f.read() == file_content diff --git a/tests/functional/api/test_projects.py b/tests/functional/api/test_projects.py index edb7e31df..2105cdcb8 100644 --- a/tests/functional/api/test_projects.py +++ b/tests/functional/api/test_projects.py @@ -401,7 +401,7 @@ def test_project_groups_list(gl, group): project = gl.projects.create(data) groups = project.groups.list() - group_ids = set([x.id for x in groups]) + group_ids = {x.id for x in groups} assert {group.id, group2.id} == group_ids diff --git a/tests/functional/api/test_repository.py b/tests/functional/api/test_repository.py index 4376c64c5..f011a1da9 100644 --- a/tests/functional/api/test_repository.py +++ b/tests/functional/api/test_repository.py @@ -1,6 +1,5 @@ import base64 import os -import sys import tarfile import time import zipfile @@ -74,9 +73,6 @@ def test_repository_archive(project): assert archive == archive2 -# NOTE(jlvillal): Support for using tarfile.is_tarfile() on a file or file-like object -# was added in Python 3.9 -@pytest.mark.skipif(sys.version_info < (3, 9), reason="requires python3.9 or higher") @pytest.mark.parametrize( "format,assertion", [ diff --git a/tests/functional/conftest.py b/tests/functional/conftest.py index 2d2815547..ae643481a 100644 --- a/tests/functional/conftest.py +++ b/tests/functional/conftest.py @@ -1,3 +1,5 @@ +from __future__ import annotations + import dataclasses import datetime import logging @@ -6,7 +8,7 @@ import time import uuid from subprocess import check_output -from typing import Optional, Sequence +from typing import Sequence import pytest import requests @@ -128,7 +130,7 @@ def set_token(container: str, fixture_dir: pathlib.Path) -> str: logging.info("Creating API token.") set_token_rb = fixture_dir / "set_token.rb" - with open(set_token_rb, "r", encoding="utf-8") as f: + with open(set_token_rb, encoding="utf-8") as f: set_token_command = f.read().strip() rails_command = [ @@ -271,7 +273,7 @@ def gl(gitlab_url: str, gitlab_token: str) -> gitlab.Gitlab: @pytest.fixture(scope="session") -def gitlab_plan(gl: gitlab.Gitlab) -> Optional[str]: +def gitlab_plan(gl: gitlab.Gitlab) -> str | None: return helpers.get_gitlab_plan(gl) diff --git a/tests/functional/helpers.py b/tests/functional/helpers.py index a898aa947..090673bf7 100644 --- a/tests/functional/helpers.py +++ b/tests/functional/helpers.py @@ -1,6 +1,8 @@ +from __future__ import annotations + import logging import time -from typing import Optional, TYPE_CHECKING +from typing import TYPE_CHECKING import pytest @@ -13,7 +15,7 @@ MAX_ITERATIONS = int(TIMEOUT / SLEEP_INTERVAL) -def get_gitlab_plan(gl: gitlab.Gitlab) -> Optional[str]: +def get_gitlab_plan(gl: gitlab.Gitlab) -> str | None: """Determine the license available on the GitLab instance""" try: license = gl.get_license() diff --git a/tests/unit/base/test_rest_object.py b/tests/unit/base/test_rest_object.py index 588c1b53e..950ff3086 100644 --- a/tests/unit/base/test_rest_object.py +++ b/tests/unit/base/test_rest_object.py @@ -1,3 +1,5 @@ +from __future__ import annotations + import pickle import pytest @@ -131,7 +133,7 @@ def test_dir_unique(fake_manager): def test_create_managers(gl, fake_manager): class ObjectWithManager(helpers.FakeObject): - fakes: "FakeManager" + fakes: FakeManager obj = ObjectWithManager(fake_manager, {"foo": "bar"}) obj.id = 42 @@ -144,7 +146,7 @@ def test_equality(fake_manager): obj1 = helpers.FakeObject(fake_manager, {"id": "foo"}) obj2 = helpers.FakeObject(fake_manager, {"id": "foo", "other_attr": "bar"}) assert obj1 == obj2 - assert len(set((obj1, obj2))) == 1 + assert len({obj1, obj2}) == 1 def test_equality_custom_id(fake_manager): @@ -169,7 +171,7 @@ def test_inequality_no_id(fake_manager): obj1 = helpers.FakeObject(fake_manager, {"attr1": "foo"}) obj2 = helpers.FakeObject(fake_manager, {"attr1": "bar"}) assert obj1 != obj2 - assert len(set((obj1, obj2))) == 2 + assert len({obj1, obj2}) == 2 def test_equality_with_other_objects(fake_manager): diff --git a/tests/unit/helpers.py b/tests/unit/helpers.py index 1093728a5..717108d44 100644 --- a/tests/unit/helpers.py +++ b/tests/unit/helpers.py @@ -1,7 +1,8 @@ +from __future__ import annotations + import datetime import io import json -from typing import Optional import requests import responses @@ -47,7 +48,7 @@ class FakeManagerWithParent(base.RESTManager): # https://github.com/patrys/httmock/ which is licensed under the Apache License, Version # 2.0. Thus it is allowed to be used in this project. # https://www.apache.org/licenses/GPL-compatibility.html -class Headers(object): +class Headers: def __init__(self, res): self.headers = res.headers @@ -64,7 +65,7 @@ def httmock_response( headers=None, reason=None, elapsed=0, - request: Optional[requests.models.PreparedRequest] = None, + request: requests.models.PreparedRequest | None = None, stream: bool = False, http_vsn=11, ) -> requests.models.Response: diff --git a/tests/unit/meta/test_imports.py b/tests/unit/meta/test_imports.py index 1f038146d..d49f3e495 100644 --- a/tests/unit/meta/test_imports.py +++ b/tests/unit/meta/test_imports.py @@ -25,7 +25,7 @@ def test_all_v4_objects_are_imported() -> None: assert len(gitlab.v4.objects.__path__) == 1 init_files: Set[str] = set() - with open(gitlab.v4.objects.__file__, "r", encoding="utf-8") as in_file: + with open(gitlab.v4.objects.__file__, encoding="utf-8") as in_file: for line in in_file.readlines(): if line.startswith("from ."): init_files.add(line.rstrip()) diff --git a/tests/unit/meta/test_mro.py b/tests/unit/meta/test_mro.py index a0f8e6068..1b64003d0 100644 --- a/tests/unit/meta/test_mro.py +++ b/tests/unit/meta/test_mro.py @@ -116,11 +116,9 @@ class definition. if mro[index_to_check].__module__ != "gitlab.base": failed_messages.append( - ( - f"class definition for {class_name!r} in file {filename!r} " - f"must have {base_classname!r} as the last class in the " - f"class definition" - ) + f"class definition for {class_name!r} in file {filename!r} " + f"must have {base_classname!r} as the last class in the " + f"class definition" ) failed_msg = "\n".join(failed_messages) assert not failed_messages, failed_msg diff --git a/tests/unit/test_cli.py b/tests/unit/test_cli.py index 5a1551ad8..c8adc1e25 100644 --- a/tests/unit/test_cli.py +++ b/tests/unit/test_cli.py @@ -2,7 +2,6 @@ import contextlib import io import os -import sys import tempfile from unittest import mock @@ -219,7 +218,6 @@ class FakeManager(CreateMixin, UpdateMixin, gitlab.base.RESTManager): ) -@pytest.mark.skipif(sys.version_info < (3, 8), reason="added in 3.8") def test_legacy_display_without_fields_warns(fake_object_no_id): printer = v4_cli.LegacyPrinter() @@ -229,7 +227,6 @@ def test_legacy_display_without_fields_warns(fake_object_no_id): assert "No default fields to show" in mocked.call_args.args[0] -@pytest.mark.skipif(sys.version_info < (3, 8), reason="added in 3.8") def test_legacy_display_with_long_repr_truncates(fake_object_long_repr): printer = v4_cli.LegacyPrinter()