Skip to content

Commit

Permalink
Better types
Browse files Browse the repository at this point in the history
  • Loading branch information
adferrand committed Aug 12, 2023
1 parent bd51277 commit 89b3ab2
Showing 1 changed file with 34 additions and 29 deletions.
63 changes: 34 additions & 29 deletions src/lexicon/client.py
@@ -1,8 +1,11 @@
"""Main module of Lexicon. Defines the Client class, that holds all Lexicon logic."""
from __future__ import annotations
import importlib
import logging
import os
from typing import Dict, List, Optional, Type, Union, cast
from contextlib import AbstractContextManager
from types import TracebackType
from typing import Type, Any

import tldextract # type: ignore

Expand All @@ -29,10 +32,10 @@ def create_record(self, rtype: str, name: str, content: str) -> bool:

def list_records(
self,
rtype: Optional[str] = None,
name: Optional[str] = None,
content: Optional[str] = None,
) -> List[Dict]:
rtype: str | None = None,
name: str | None = None,
content: str | None = None,
) -> list[dict[str, Any]]:
"""
List all records. Return an empty list if no records found
type, name and content are used to filter records.
Expand All @@ -42,10 +45,10 @@ def list_records(

def update_record(
self,
identifier: Optional[str] = None,
rtype: Optional[str] = None,
name: Optional[str] = None,
content: Optional[str] = None,
identifier: str | None = None,
rtype: str | None = None,
name: str | None = None,
content: str | None = None,
) -> bool:
"""
Update a record. Identifier must be specified.
Expand All @@ -54,10 +57,10 @@ def update_record(

def delete_record(
self,
identifier: Optional[str] = None,
rtype: Optional[str] = None,
name: Optional[str] = None,
content: Optional[str] = None,
identifier: str | None = None,
rtype: str | None = None,
name: str | None = None,
content: str | None = None,
) -> bool:
"""
Delete an existing record.
Expand All @@ -67,14 +70,14 @@ def delete_record(
return self.provider.delete_record(identifier, rtype, name, content)


class Client:
class Client(AbstractContextManager):
"""This is the Lexicon client, that will execute all the logic."""

def __init__(
self, config: Optional[Union[helper_config.ConfigResolver, Dict]] = None
self, config: helper_config.ConfigResolver | dict[str, Any] | None = None
):
if not config:
# If there is not config specified, we load a non-interactive configuration.
# If there is no config specified, we load a non-interactive configuration.
self.config = helper_config.non_interactive_config_resolver()
elif not isinstance(config, helper_config.ConfigResolver):
# If config is not a ConfigResolver, we are in a legacy situation.
Expand All @@ -83,8 +86,11 @@ def __init__(
else:
self.config = config

# Validate configuration
self._validate_config()
domain = self.config.resolve("lexicon:domain")
if not domain:
raise AttributeError("domain")

self._validate_provider()

runtime_config = {}

Expand All @@ -97,9 +103,7 @@ def __init__(
domain_extractor = tldextract.TLDExtract(
cache_file=_resolve_tldextract_cache_path(), include_psl_private_domains=True # type: ignore
)
domain_parts = domain_extractor(
cast(str, self.config.resolve("lexicon:domain"))
)
domain_parts = domain_extractor(domain)
runtime_config["domain"] = f"{domain_parts.domain}.{domain_parts.suffix}"

delegated = self.config.resolve("lexicon:delegated")
Expand Down Expand Up @@ -128,19 +132,23 @@ def __init__(
"lexicon.providers." + self.provider_name
)
self.provider_class: Type[Provider] = getattr(provider_module, "Provider")
self._provider: Optional[Provider]
self._provider: Provider | None

def __enter__(self) -> "_ClientExecutor":
def __enter__(self) -> _ClientExecutor:
self._provider = self.provider_class(self.config)
self._provider.authenticate()

return _ClientExecutor(self._provider)

def __exit__(self, exc_type, exc_value, traceback) -> None:
def __exit__(self, __exc_type: type[BaseException] | None, __exc_value: BaseException | None,
__traceback: TracebackType | None) -> bool | None:
if self._provider:
self._provider.cleanup()
self._provider = None

def execute(self) -> Union[bool, List[Dict]]:
return None

def execute(self) -> bool | list[dict[str, Any]]:
"""Execute provided configuration in class constructor to the DNS records"""
action = self.config.resolve("lexicon:action")
identifier = self.config.resolve("lexicon:identifier")
Expand Down Expand Up @@ -174,7 +182,7 @@ def execute(self) -> Union[bool, List[Dict]]:
finally:
self.__exit__(None, None, None)

def _validate_config(self) -> None:
def _validate_provider(self) -> None:
provider_name = self.config.resolve("lexicon:provider_name")
if not provider_name:
raise AttributeError("provider_name")
Expand All @@ -192,9 +200,6 @@ def _validate_config(self) -> None:
f"Please run `pip install dns-lexicon[{provider_name}]` first before using it."
)

if not self.config.resolve("lexicon:domain"):
raise AttributeError("domain")


def _resolve_tldextract_cache_path() -> str:
if os.environ.get("TLDEXTRACT_CACHE_FILE"):
Expand Down

0 comments on commit 89b3ab2

Please sign in to comment.