types: CredentialsConfiguration.conf can return None

This commit is contained in:
Alex Zorin
2023-04-16 10:43:00 +10:00
parent 4740e20725
commit 5503d12395
11 changed files with 34 additions and 24 deletions
@@ -2,6 +2,7 @@
import logging import logging
from typing import Any from typing import Any
from typing import Callable from typing import Callable
from typing import cast
from typing import Optional from typing import Optional
import digitalocean import digitalocean
@@ -56,7 +57,7 @@ class Authenticator(dns_common.DNSAuthenticator):
def _get_digitalocean_client(self) -> "_DigitalOceanClient": def _get_digitalocean_client(self) -> "_DigitalOceanClient":
if not self.credentials: # pragma: no cover if not self.credentials: # pragma: no cover
raise errors.Error("Plugin has not been prepared.") raise errors.Error("Plugin has not been prepared.")
return _DigitalOceanClient(self.credentials.conf('token')) return _DigitalOceanClient(cast(str, self.credentials.conf('token')))
class _DigitalOceanClient: class _DigitalOceanClient:
@@ -2,6 +2,7 @@
import logging import logging
from typing import Any from typing import Any
from typing import Callable from typing import Callable
from typing import cast
from typing import Optional from typing import Optional
from lexicon.providers import dnsimple from lexicon.providers import dnsimple
@@ -58,7 +59,7 @@ class Authenticator(dns_common.DNSAuthenticator):
def _get_dnsimple_client(self) -> "_DNSimpleLexiconClient": def _get_dnsimple_client(self) -> "_DNSimpleLexiconClient":
if not self.credentials: # pragma: no cover if not self.credentials: # pragma: no cover
raise errors.Error("Plugin has not been prepared.") raise errors.Error("Plugin has not been prepared.")
return _DNSimpleLexiconClient(self.credentials.conf('token'), self.ttl) return _DNSimpleLexiconClient(cast(str, self.credentials.conf('token')), self.ttl)
class _DNSimpleLexiconClient(dns_common_lexicon.LexiconClient): class _DNSimpleLexiconClient(dns_common_lexicon.LexiconClient):
@@ -2,6 +2,7 @@
import logging import logging
from typing import Any from typing import Any
from typing import Callable from typing import Callable
from typing import cast
from typing import Optional from typing import Optional
from lexicon.providers import dnsmadeeasy from lexicon.providers import dnsmadeeasy
@@ -62,8 +63,8 @@ class Authenticator(dns_common.DNSAuthenticator):
def _get_dnsmadeeasy_client(self) -> "_DNSMadeEasyLexiconClient": def _get_dnsmadeeasy_client(self) -> "_DNSMadeEasyLexiconClient":
if not self.credentials: # pragma: no cover if not self.credentials: # pragma: no cover
raise errors.Error("Plugin has not been prepared.") raise errors.Error("Plugin has not been prepared.")
return _DNSMadeEasyLexiconClient(self.credentials.conf('api-key'), return _DNSMadeEasyLexiconClient(cast(str,self.credentials.conf('api-key')),
self.credentials.conf('secret-key'), cast(str,self.credentials.conf('secret-key')),
self.ttl) self.ttl)
@@ -2,6 +2,7 @@
import logging import logging
from typing import Any from typing import Any
from typing import Callable from typing import Callable
from typing import cast
from typing import Optional from typing import Optional
from lexicon.providers import gehirn from lexicon.providers import gehirn
@@ -64,8 +65,8 @@ class Authenticator(dns_common.DNSAuthenticator):
if not self.credentials: # pragma: no cover if not self.credentials: # pragma: no cover
raise errors.Error("Plugin has not been prepared.") raise errors.Error("Plugin has not been prepared.")
return _GehirnLexiconClient( return _GehirnLexiconClient(
self.credentials.conf('api-token'), cast(str,self.credentials.conf('api-token')),
self.credentials.conf('api-secret'), cast(str,self.credentials.conf('api-secret')),
self.ttl self.ttl
) )
@@ -3,6 +3,7 @@ import logging
import re import re
from typing import Any from typing import Any
from typing import Callable from typing import Callable
from typing import cast
from typing import Optional from typing import Optional
from typing import Union from typing import Union
@@ -61,7 +62,7 @@ class Authenticator(dns_common.DNSAuthenticator):
def _get_linode_client(self) -> '_LinodeLexiconClient': def _get_linode_client(self) -> '_LinodeLexiconClient':
if not self.credentials: # pragma: no cover if not self.credentials: # pragma: no cover
raise errors.Error("Plugin has not been prepared.") raise errors.Error("Plugin has not been prepared.")
api_key = self.credentials.conf('key') api_key = cast(str,self.credentials.conf('key'))
api_version: Optional[Union[str, int]] = self.credentials.conf('version') api_version: Optional[Union[str, int]] = self.credentials.conf('version')
if api_version == '': if api_version == '':
api_version = None api_version = None
@@ -2,6 +2,7 @@
import logging import logging
from typing import Any from typing import Any
from typing import Callable from typing import Callable
from typing import cast
from typing import Optional from typing import Optional
from lexicon.providers import luadns from lexicon.providers import luadns
@@ -59,8 +60,8 @@ class Authenticator(dns_common.DNSAuthenticator):
def _get_luadns_client(self) -> "_LuaDNSLexiconClient": def _get_luadns_client(self) -> "_LuaDNSLexiconClient":
if not self.credentials: # pragma: no cover if not self.credentials: # pragma: no cover
raise errors.Error("Plugin has not been prepared.") raise errors.Error("Plugin has not been prepared.")
return _LuaDNSLexiconClient(self.credentials.conf('email'), return _LuaDNSLexiconClient(cast(str,self.credentials.conf('email')),
self.credentials.conf('token'), cast(str,self.credentials.conf('token')),
self.ttl) self.ttl)
@@ -2,6 +2,7 @@
import logging import logging
from typing import Any from typing import Any
from typing import Callable from typing import Callable
from typing import cast
from typing import Optional from typing import Optional
from lexicon.providers import nsone from lexicon.providers import nsone
@@ -58,7 +59,7 @@ class Authenticator(dns_common.DNSAuthenticator):
def _get_nsone_client(self) -> "_NS1LexiconClient": def _get_nsone_client(self) -> "_NS1LexiconClient":
if not self.credentials: # pragma: no cover if not self.credentials: # pragma: no cover
raise errors.Error("Plugin has not been prepared.") raise errors.Error("Plugin has not been prepared.")
return _NS1LexiconClient(self.credentials.conf('api-key'), self.ttl) return _NS1LexiconClient(cast(str, self.credentials.conf('api-key')), self.ttl)
class _NS1LexiconClient(dns_common_lexicon.LexiconClient): class _NS1LexiconClient(dns_common_lexicon.LexiconClient):
@@ -2,6 +2,7 @@
import logging import logging
from typing import Any from typing import Any
from typing import Callable from typing import Callable
from typing import cast
from typing import Optional from typing import Optional
from lexicon.providers import ovh from lexicon.providers import ovh
@@ -65,10 +66,10 @@ class Authenticator(dns_common.DNSAuthenticator):
if not self.credentials: # pragma: no cover if not self.credentials: # pragma: no cover
raise errors.Error("Plugin has not been prepared.") raise errors.Error("Plugin has not been prepared.")
return _OVHLexiconClient( return _OVHLexiconClient(
self.credentials.conf('endpoint'), cast(str, self.credentials.conf('endpoint')),
self.credentials.conf('application-key'), cast(str,self.credentials.conf('application-key')),
self.credentials.conf('application-secret'), cast(str,self.credentials.conf('application-secret')),
self.credentials.conf('consumer-key'), cast(str,self.credentials.conf('consumer-key')),
self.ttl self.ttl
) )
@@ -2,6 +2,7 @@
import logging import logging
from typing import Any from typing import Any
from typing import Callable from typing import Callable
from typing import cast
from typing import Optional from typing import Optional
import dns.flags import dns.flags
@@ -59,7 +60,7 @@ class Authenticator(dns_common.DNSAuthenticator):
'RFC 2136 Dynamic Updates.' 'RFC 2136 Dynamic Updates.'
def _validate_credentials(self, credentials: CredentialsConfiguration) -> None: def _validate_credentials(self, credentials: CredentialsConfiguration) -> None:
server = credentials.conf('server') server = cast(str, credentials.conf('server'))
if not is_ipaddress(server): if not is_ipaddress(server):
raise errors.PluginError("The configured target DNS server ({0}) is not a valid IPv4 " raise errors.PluginError("The configured target DNS server ({0}) is not a valid IPv4 "
"or IPv6 address. A hostname is not allowed.".format(server)) "or IPv6 address. A hostname is not allowed.".format(server))
@@ -89,11 +90,11 @@ class Authenticator(dns_common.DNSAuthenticator):
def _get_rfc2136_client(self) -> "_RFC2136Client": def _get_rfc2136_client(self) -> "_RFC2136Client":
if not self.credentials: # pragma: no cover if not self.credentials: # pragma: no cover
raise errors.Error("Plugin has not been prepared.") raise errors.Error("Plugin has not been prepared.")
return _RFC2136Client(self.credentials.conf('server'), return _RFC2136Client(cast(str, self.credentials.conf('server')),
int(self.credentials.conf('port') or self.PORT), int(cast(str, self.credentials.conf('port')) or self.PORT),
self.credentials.conf('name'), cast(str, self.credentials.conf('name')),
self.credentials.conf('secret'), cast(str, self.credentials.conf('secret')),
self.ALGORITHMS.get(self.credentials.conf('algorithm'), self.ALGORITHMS.get(self.credentials.conf('algorithm') or '',
dns.tsig.HMAC_MD5)) dns.tsig.HMAC_MD5))
@@ -2,6 +2,7 @@
import logging import logging
from typing import Any from typing import Any
from typing import Callable from typing import Callable
from typing import cast
from typing import Optional from typing import Optional
from lexicon.providers import sakuracloud from lexicon.providers import sakuracloud
@@ -64,8 +65,8 @@ class Authenticator(dns_common.DNSAuthenticator):
if not self.credentials: # pragma: no cover if not self.credentials: # pragma: no cover
raise errors.Error("Plugin has not been prepared.") raise errors.Error("Plugin has not been prepared.")
return _SakuraCloudLexiconClient( return _SakuraCloudLexiconClient(
self.credentials.conf('api-token'), cast(str, self.credentials.conf('api-token')),
self.credentials.conf('api-secret'), cast(str, self.credentials.conf('api-secret')),
self.ttl self.ttl
) )
+2 -2
View File
@@ -312,7 +312,7 @@ class CredentialsConfiguration:
) )
) )
def conf(self, var: str) -> str: def conf(self, var: str) -> Optional[str]:
"""Find a configuration value for variable `var`, as transformed by `mapper`. """Find a configuration value for variable `var`, as transformed by `mapper`.
:param str var: The variable to get. :param str var: The variable to get.
@@ -325,7 +325,7 @@ class CredentialsConfiguration:
def _has(self, var: str) -> bool: def _has(self, var: str) -> bool:
return self.mapper(var) in self.confobj return self.mapper(var) in self.confobj
def _get(self, var: str) -> str: def _get(self, var: str) -> Optional[str]:
return self.confobj.get(self.mapper(var)) return self.confobj.get(self.mapper(var))