*: replace List from typing with normal list
This commit is contained in:
parent
fdde97cbbf
commit
ecff810021
|
@ -2,7 +2,6 @@
|
|||
import time
|
||||
from collections import Counter
|
||||
from datetime import timedelta
|
||||
from typing import List
|
||||
|
||||
from django.db.models import Count, ExpressionWrapper, F, Model
|
||||
from django.db.models.fields import DurationField
|
||||
|
@ -19,7 +18,7 @@ from rest_framework.viewsets import ViewSet
|
|||
from authentik.events.models import Event, EventAction
|
||||
|
||||
|
||||
def get_events_per_1h(**filter_kwargs) -> List[dict[str, int]]:
|
||||
def get_events_per_1h(**filter_kwargs) -> list[dict[str, int]]:
|
||||
"""Get event count by hour in the last day, fill with zeros"""
|
||||
date_from = now() - timedelta(days=1)
|
||||
result = (
|
||||
|
|
|
@ -1,5 +1,5 @@
|
|||
"""authentik admin util views"""
|
||||
from typing import Any, List, Optional
|
||||
from typing import Any, Optional
|
||||
from urllib.parse import urlparse
|
||||
|
||||
from django.contrib import messages
|
||||
|
@ -40,7 +40,7 @@ class SearchListMixin(MultipleObjectMixin):
|
|||
"""Accept search query using `search` querystring parameter. Requires self.search_fields,
|
||||
a list of all fields to search. Can contain special lookups like __icontains"""
|
||||
|
||||
search_fields: List[str]
|
||||
search_fields: list[str]
|
||||
|
||||
def get_queryset(self) -> QuerySet:
|
||||
queryset = super().get_queryset()
|
||||
|
|
|
@ -3,7 +3,7 @@ from dataclasses import dataclass, field
|
|||
from datetime import datetime
|
||||
from enum import Enum
|
||||
from traceback import format_tb
|
||||
from typing import Any, List, Optional
|
||||
from typing import Any, Optional
|
||||
|
||||
from celery import Task
|
||||
from django.core.cache import cache
|
||||
|
@ -26,7 +26,7 @@ class TaskResult:
|
|||
|
||||
status: TaskResultStatus
|
||||
|
||||
messages: List[str] = field(default_factory=list)
|
||||
messages: list[str] = field(default_factory=list)
|
||||
|
||||
# Optional UID used in cache for tasks that run in different instances
|
||||
uid: Optional[str] = field(default=None)
|
||||
|
@ -49,7 +49,7 @@ class TaskInfo:
|
|||
|
||||
task_call_module: str
|
||||
task_call_func: str
|
||||
task_call_args: List[Any] = field(default_factory=list)
|
||||
task_call_args: list[Any] = field(default_factory=list)
|
||||
task_call_kwargs: dict[str, Any] = field(default_factory=dict)
|
||||
|
||||
task_description: Optional[str] = field(default=None)
|
||||
|
@ -109,7 +109,7 @@ class MonitoredTask(Task):
|
|||
|
||||
# pylint: disable=too-many-arguments
|
||||
def after_return(
|
||||
self, status, retval, task_id, args: List[Any], kwargs: dict[str, Any], einfo
|
||||
self, status, retval, task_id, args: list[Any], kwargs: dict[str, Any], einfo
|
||||
):
|
||||
if not self._result.uid:
|
||||
self._result.uid = self._uid
|
||||
|
|
|
@ -1,6 +1,6 @@
|
|||
"""Flows Planner"""
|
||||
from dataclasses import dataclass, field
|
||||
from typing import Any, List, Optional
|
||||
from typing import Any, Optional
|
||||
|
||||
from django.core.cache import cache
|
||||
from django.http import HttpRequest
|
||||
|
@ -38,9 +38,9 @@ class FlowPlan:
|
|||
|
||||
flow_pk: str
|
||||
|
||||
stages: List[Stage] = field(default_factory=list)
|
||||
stages: list[Stage] = field(default_factory=list)
|
||||
context: dict[str, Any] = field(default_factory=dict)
|
||||
markers: List[StageMarker] = field(default_factory=list)
|
||||
markers: list[StageMarker] = field(default_factory=list)
|
||||
|
||||
def append(self, stage: Stage, marker: Optional[StageMarker] = None):
|
||||
"""Append `stage` to all stages, optionall with stage marker"""
|
||||
|
|
|
@ -1,6 +1,6 @@
|
|||
"""transfer common classes"""
|
||||
from dataclasses import asdict, dataclass, field, is_dataclass
|
||||
from typing import Any, List
|
||||
from typing import Any
|
||||
from uuid import UUID
|
||||
|
||||
from django.core.serializers.json import DjangoJSONEncoder
|
||||
|
@ -61,7 +61,7 @@ class FlowBundle:
|
|||
"""Dataclass used for a full export"""
|
||||
|
||||
version: int = field(default=1)
|
||||
entries: List[FlowBundleEntry] = field(default_factory=list)
|
||||
entries: list[FlowBundleEntry] = field(default_factory=list)
|
||||
|
||||
|
||||
class DataclassEncoder(DjangoJSONEncoder):
|
||||
|
|
|
@ -1,6 +1,6 @@
|
|||
"""Flow exporter"""
|
||||
from json import dumps
|
||||
from typing import Iterator, List
|
||||
from typing import Iterator
|
||||
from uuid import UUID
|
||||
|
||||
from django.db.models import Q
|
||||
|
@ -22,7 +22,7 @@ class FlowExporter:
|
|||
with_policies: bool
|
||||
with_stage_prompts: bool
|
||||
|
||||
pbm_uuids: List[UUID]
|
||||
pbm_uuids: list[UUID]
|
||||
|
||||
def __init__(self, flow: Flow):
|
||||
self.flow = flow
|
||||
|
|
|
@ -1,8 +1,8 @@
|
|||
"""authentik UI utils"""
|
||||
from typing import Any, List
|
||||
from typing import Any
|
||||
|
||||
|
||||
def human_list(_list: List[Any]) -> str:
|
||||
def human_list(_list: list[Any]) -> str:
|
||||
"""Convert a list of items into 'a, b or c'"""
|
||||
last_item = _list.pop()
|
||||
if len(_list) < 1:
|
||||
|
|
|
@ -1,6 +1,6 @@
|
|||
"""Kubernetes deployment controller"""
|
||||
from io import StringIO
|
||||
from typing import List, Type
|
||||
from typing import Type
|
||||
|
||||
from kubernetes.client import OpenApiException
|
||||
from kubernetes.client.api_client import ApiClient
|
||||
|
@ -19,7 +19,7 @@ class KubernetesController(BaseController):
|
|||
"""Manage deployment of outpost in kubernetes"""
|
||||
|
||||
reconcilers: dict[str, Type[KubernetesObjectReconciler]]
|
||||
reconcile_order: List[str]
|
||||
reconcile_order: list[str]
|
||||
|
||||
client: ApiClient
|
||||
connection: KubernetesServiceConnection
|
||||
|
@ -45,7 +45,7 @@ class KubernetesController(BaseController):
|
|||
except OpenApiException as exc:
|
||||
raise ControllerException from exc
|
||||
|
||||
def up_with_logs(self) -> List[str]:
|
||||
def up_with_logs(self) -> list[str]:
|
||||
try:
|
||||
all_logs = []
|
||||
for reconcile_key in self.reconcile_order:
|
||||
|
|
|
@ -1,7 +1,7 @@
|
|||
"""Outpost models"""
|
||||
from dataclasses import asdict, dataclass, field
|
||||
from datetime import datetime
|
||||
from typing import Iterable, List, Optional, Type, Union
|
||||
from typing import Iterable, Optional, Type, Union
|
||||
from uuid import uuid4
|
||||
|
||||
from dacite import from_dict
|
||||
|
@ -315,7 +315,7 @@ class Outpost(models.Model):
|
|||
return f"outpost_{self.uuid.hex}_state"
|
||||
|
||||
@property
|
||||
def state(self) -> List["OutpostState"]:
|
||||
def state(self) -> list["OutpostState"]:
|
||||
"""Get outpost's health status"""
|
||||
return OutpostState.for_outpost(self)
|
||||
|
||||
|
@ -399,7 +399,7 @@ class OutpostState:
|
|||
return parse(self.version) < OUR_VERSION
|
||||
|
||||
@staticmethod
|
||||
def for_outpost(outpost: Outpost) -> List["OutpostState"]:
|
||||
def for_outpost(outpost: Outpost) -> list["OutpostState"]:
|
||||
"""Get all states for an outpost"""
|
||||
keys = cache.keys(f"{outpost.state_cache_prefix}_*")
|
||||
states = []
|
||||
|
|
|
@ -2,7 +2,7 @@
|
|||
from enum import Enum
|
||||
from multiprocessing import Pipe, current_process
|
||||
from multiprocessing.connection import Connection
|
||||
from typing import Iterator, List, Optional
|
||||
from typing import Iterator, Optional
|
||||
|
||||
from django.core.cache import cache
|
||||
from django.http import HttpRequest
|
||||
|
@ -54,8 +54,8 @@ class PolicyEngine:
|
|||
empty_result: bool
|
||||
|
||||
__pbm: PolicyBindingModel
|
||||
__cached_policies: List[PolicyResult]
|
||||
__processes: List[PolicyProcessInfo]
|
||||
__cached_policies: list[PolicyResult]
|
||||
__processes: list[PolicyProcessInfo]
|
||||
|
||||
__expected_result_count: int
|
||||
|
||||
|
@ -137,7 +137,7 @@ class PolicyEngine:
|
|||
@property
|
||||
def result(self) -> PolicyResult:
|
||||
"""Get policy-checking result"""
|
||||
process_results: List[PolicyResult] = [
|
||||
process_results: list[PolicyResult] = [
|
||||
x.result for x in self.__processes if x.result
|
||||
]
|
||||
all_results = list(process_results + self.__cached_policies)
|
||||
|
|
|
@ -1,6 +1,6 @@
|
|||
"""authentik expression policy evaluator"""
|
||||
from ipaddress import ip_address, ip_network
|
||||
from typing import TYPE_CHECKING, List, Optional
|
||||
from typing import TYPE_CHECKING, Optional
|
||||
|
||||
from django.http import HttpRequest
|
||||
from structlog.stdlib import get_logger
|
||||
|
@ -19,7 +19,7 @@ if TYPE_CHECKING:
|
|||
class PolicyEvaluator(BaseEvaluator):
|
||||
"""Validate and evaluate python-based expressions"""
|
||||
|
||||
_messages: List[str]
|
||||
_messages: list[str]
|
||||
|
||||
policy: Optional["ExpressionPolicy"] = None
|
||||
|
||||
|
|
|
@ -5,7 +5,7 @@ import json
|
|||
import time
|
||||
from dataclasses import asdict, dataclass, field
|
||||
from hashlib import sha256
|
||||
from typing import Any, List, Optional, Type
|
||||
from typing import Any, Optional, Type
|
||||
from urllib.parse import urlparse
|
||||
from uuid import uuid4
|
||||
|
||||
|
@ -218,7 +218,7 @@ class OAuth2Provider(Provider):
|
|||
)
|
||||
|
||||
def create_refresh_token(
|
||||
self, user: User, scope: List[str], request: HttpRequest
|
||||
self, user: User, scope: list[str], request: HttpRequest
|
||||
) -> "RefreshToken":
|
||||
"""Create and populate a RefreshToken object."""
|
||||
token = RefreshToken(
|
||||
|
@ -231,7 +231,7 @@ class OAuth2Provider(Provider):
|
|||
token.access_token = token.create_access_token(user, request)
|
||||
return token
|
||||
|
||||
def get_jwt_keys(self) -> List[Key]:
|
||||
def get_jwt_keys(self) -> list[Key]:
|
||||
"""
|
||||
Takes a provider and returns the set of keys associated with it.
|
||||
Returns a list of keys.
|
||||
|
@ -321,7 +321,7 @@ class BaseGrantModel(models.Model):
|
|||
_scope = models.TextField(default="", verbose_name=_("Scopes"))
|
||||
|
||||
@property
|
||||
def scope(self) -> List[str]:
|
||||
def scope(self) -> list[str]:
|
||||
"""Return scopes as list of strings"""
|
||||
return self._scope.split()
|
||||
|
||||
|
|
|
@ -2,7 +2,7 @@
|
|||
import re
|
||||
from base64 import b64decode
|
||||
from binascii import Error
|
||||
from typing import List, Optional, Tuple
|
||||
from typing import Optional, Tuple
|
||||
|
||||
from django.http import HttpRequest, HttpResponse, JsonResponse
|
||||
from django.utils.cache import patch_vary_headers
|
||||
|
@ -92,7 +92,7 @@ def extract_client_auth(request: HttpRequest) -> Tuple[str, str]:
|
|||
return (client_id, client_secret)
|
||||
|
||||
|
||||
def protected_resource_view(scopes: List[str]):
|
||||
def protected_resource_view(scopes: list[str]):
|
||||
"""View decorator. The client accesses protected resources by presenting the
|
||||
access token to the resource server.
|
||||
|
||||
|
|
|
@ -1,7 +1,7 @@
|
|||
"""authentik OAuth2 Authorization views"""
|
||||
from dataclasses import dataclass, field
|
||||
from datetime import timedelta
|
||||
from typing import List, Optional, Set
|
||||
from typing import Optional, Set
|
||||
from urllib.parse import parse_qs, urlencode, urlsplit, urlunsplit
|
||||
from uuid import uuid4
|
||||
|
||||
|
@ -69,7 +69,7 @@ class OAuthAuthorizationParams:
|
|||
client_id: str
|
||||
redirect_uri: str
|
||||
response_type: str
|
||||
scope: List[str]
|
||||
scope: list[str]
|
||||
state: str
|
||||
nonce: Optional[str]
|
||||
prompt: Set[str]
|
||||
|
|
|
@ -2,7 +2,7 @@
|
|||
from base64 import urlsafe_b64encode
|
||||
from dataclasses import InitVar, dataclass
|
||||
from hashlib import sha256
|
||||
from typing import Any, List, Optional
|
||||
from typing import Any, Optional
|
||||
|
||||
from django.http import HttpRequest, HttpResponse
|
||||
from django.views import View
|
||||
|
@ -33,7 +33,7 @@ class TokenParams:
|
|||
redirect_uri: str
|
||||
grant_type: str
|
||||
state: str
|
||||
scope: List[str]
|
||||
scope: list[str]
|
||||
|
||||
authorization_code: Optional[AuthorizationCode] = None
|
||||
refresh_token: Optional[RefreshToken] = None
|
||||
|
|
|
@ -1,5 +1,5 @@
|
|||
"""authentik OAuth2 OpenID Userinfo views"""
|
||||
from typing import Any, List
|
||||
from typing import Any
|
||||
|
||||
from django.http import HttpRequest, HttpResponse
|
||||
from django.utils.translation import gettext_lazy as _
|
||||
|
@ -22,7 +22,7 @@ class UserInfoView(View):
|
|||
"""Create a dictionary with all the requested claims about the End-User.
|
||||
See: http://openid.net/specs/openid-connect-core-1_0.html#UserInfoResponse"""
|
||||
|
||||
def get_scope_descriptions(self, scopes: List[str]) -> dict[str, str]:
|
||||
def get_scope_descriptions(self, scopes: list[str]) -> dict[str, str]:
|
||||
"""Get a list of all Scopes's descriptions"""
|
||||
scope_descriptions = {}
|
||||
for scope in ScopeMapping.objects.filter(scope_name__in=scopes).order_by(
|
||||
|
|
|
@ -1,6 +1,6 @@
|
|||
"""Source type manager"""
|
||||
from enum import Enum
|
||||
from typing import Callable, List
|
||||
from typing import Callable
|
||||
|
||||
from django.utils.text import slugify
|
||||
from structlog.stdlib import get_logger
|
||||
|
@ -23,7 +23,7 @@ class SourceTypeManager:
|
|||
"""Manager to hold all Source types."""
|
||||
|
||||
__source_types: dict[RequestKind, dict[str, Callable]] = {}
|
||||
__names: List[str] = []
|
||||
__names: list[str] = []
|
||||
|
||||
def source(self, kind: RequestKind, name: str):
|
||||
"""Class decorator to register classes inline."""
|
||||
|
|
|
@ -1,5 +1,5 @@
|
|||
"""authentik consent stage"""
|
||||
from typing import Any, List
|
||||
from typing import Any
|
||||
|
||||
from django.http import HttpRequest, HttpResponse
|
||||
from django.utils.timezone import now
|
||||
|
@ -25,7 +25,7 @@ class ConsentStageView(FormView, StageView):
|
|||
kwargs["context"] = self.executor.plan.context
|
||||
return kwargs
|
||||
|
||||
def get_template_names(self) -> List[str]:
|
||||
def get_template_names(self) -> list[str]:
|
||||
# PLAN_CONTEXT_CONSENT_TEMPLATE has to be set by a template that calls this stage
|
||||
if PLAN_CONTEXT_CONSENT_TEMPLATE in self.executor.plan.context:
|
||||
template_name = self.executor.plan.context[PLAN_CONTEXT_CONSENT_TEMPLATE]
|
||||
|
|
|
@ -1,5 +1,5 @@
|
|||
"""Identification stage logic"""
|
||||
from typing import List, Optional
|
||||
from typing import Optional
|
||||
|
||||
from django.contrib import messages
|
||||
from django.db.models import Q
|
||||
|
@ -29,7 +29,7 @@ class IdentificationStageView(FormView, StageView):
|
|||
kwargs["stage"] = self.executor.current_stage
|
||||
return kwargs
|
||||
|
||||
def get_template_names(self) -> List[str]:
|
||||
def get_template_names(self) -> list[str]:
|
||||
current_stage: IdentificationStage = self.executor.current_stage
|
||||
return [current_stage.template]
|
||||
|
||||
|
@ -56,7 +56,7 @@ class IdentificationStageView(FormView, StageView):
|
|||
|
||||
# Check all enabled source, add them if they have a UI Login button.
|
||||
kwargs["sources"] = []
|
||||
sources: List[Source] = (
|
||||
sources: list[Source] = (
|
||||
Source.objects.filter(enabled=True).order_by("name").select_subclasses()
|
||||
)
|
||||
for source in sources:
|
||||
|
|
|
@ -1,5 +1,5 @@
|
|||
"""authentik password stage"""
|
||||
from typing import Any, List, Optional
|
||||
from typing import Any, Optional
|
||||
|
||||
from django.contrib.auth import _clean_credentials
|
||||
from django.contrib.auth.backends import BaseBackend
|
||||
|
@ -24,7 +24,7 @@ SESSION_INVALID_TRIES = "user_invalid_tries"
|
|||
|
||||
|
||||
def authenticate(
|
||||
request: HttpRequest, backends: List[str], **credentials: dict[str, Any]
|
||||
request: HttpRequest, backends: list[str], **credentials: dict[str, Any]
|
||||
) -> Optional[User]:
|
||||
"""If the given credentials are valid, return a User object.
|
||||
|
||||
|
|
|
@ -1,7 +1,7 @@
|
|||
"""Prompt forms"""
|
||||
from email.policy import Policy
|
||||
from types import MethodType
|
||||
from typing import Any, Callable, Iterator, List
|
||||
from typing import Any, Callable, Iterator
|
||||
|
||||
from django import forms
|
||||
from django.db.models.query import QuerySet
|
||||
|
@ -52,10 +52,10 @@ class PromptAdminForm(forms.ModelForm):
|
|||
class ListPolicyEngine(PolicyEngine):
|
||||
"""Slightly modified policy engine, which uses a list instead of a PolicyBindingModel"""
|
||||
|
||||
__list: List[Policy]
|
||||
__list: list[Policy]
|
||||
|
||||
def __init__(
|
||||
self, policies: List[Policy], user: User, request: HttpRequest = None
|
||||
self, policies: list[Policy], user: User, request: HttpRequest = None
|
||||
) -> None:
|
||||
super().__init__(PolicyBindingModel(), user, request)
|
||||
self.__list = policies
|
||||
|
|
Reference in a new issue