mirror of
https://github.com/cisagov/manage.get.gov.git
synced 2025-07-22 10:46:06 +02:00
formatted code for linter; reformatted _fetch_cache for readability
This commit is contained in:
parent
c5afb85473
commit
74e2f7117b
2 changed files with 50 additions and 45 deletions
|
@ -1601,50 +1601,11 @@ class Domain(TimeStampedModel, DomainHelper):
|
|||
def _fetch_cache(self, fetch_hosts=False, fetch_contacts=False):
|
||||
"""Contact registry for info about a domain."""
|
||||
try:
|
||||
# get info from registry
|
||||
data_response = self._get_or_create_domain()
|
||||
cache = self._extract_data_from_response(data_response)
|
||||
|
||||
# remove null properties (to distinguish between "a value of None" and null)
|
||||
cleaned = self._remove_null_properties(cache)
|
||||
|
||||
if "statuses" in cleaned:
|
||||
cleaned["statuses"] = [status.state for status in cleaned["statuses"]]
|
||||
|
||||
cleaned["dnssecdata"] = self._get_dnssec_data(data_response.extensions)
|
||||
|
||||
# Capture and store old hosts and contacts from cache if they exist
|
||||
old_cache_hosts = self._cache.get("hosts")
|
||||
old_cache_contacts = self._cache.get("contacts")
|
||||
|
||||
if fetch_contacts:
|
||||
cleaned["contacts"] = self._get_contacts(cleaned.get("_contacts", []))
|
||||
if old_cache_hosts is not None:
|
||||
logger.debug("resetting cleaned['hosts'] to old_cache_hosts")
|
||||
cleaned["hosts"] = old_cache_hosts
|
||||
|
||||
if fetch_hosts:
|
||||
cleaned["hosts"] = self._get_hosts(cleaned.get("_hosts", []))
|
||||
if old_cache_contacts is not None:
|
||||
cleaned["contacts"] = old_cache_contacts
|
||||
|
||||
requires_save = False
|
||||
|
||||
# if expiration date from registry does not match what is in db,
|
||||
# update the db
|
||||
if "ex_date" in cleaned and cleaned["ex_date"] != self.expiration_date:
|
||||
self.expiration_date = cleaned["ex_date"]
|
||||
requires_save = True
|
||||
|
||||
# if creation_date from registry does not match what is in db,
|
||||
# update the db
|
||||
if "cr_date" in cleaned and cleaned["cr_date"] != self.created_at:
|
||||
self.created_at = cleaned["cr_date"]
|
||||
requires_save = True
|
||||
|
||||
# if either registration date or creation date need updating
|
||||
if requires_save:
|
||||
self.save()
|
||||
cleaned = self._clean_cache(cache, data_response)
|
||||
self._update_hosts_and_contacts(cleaned, fetch_hosts, fetch_contacts)
|
||||
self._update_dates(cleaned)
|
||||
|
||||
self._cache = cleaned
|
||||
|
||||
|
@ -1652,6 +1613,7 @@ class Domain(TimeStampedModel, DomainHelper):
|
|||
logger.error(e)
|
||||
|
||||
def _extract_data_from_response(self, data_response):
|
||||
"""extract datea from response from registry"""
|
||||
data = data_response.res_data[0]
|
||||
return {
|
||||
"auth_info": getattr(data, "auth_info", ...),
|
||||
|
@ -1666,6 +1628,15 @@ class Domain(TimeStampedModel, DomainHelper):
|
|||
"up_date": getattr(data, "up_date", ...),
|
||||
}
|
||||
|
||||
def _clean_cache(self, cache, data_response):
|
||||
"""clean up the cache"""
|
||||
# remove null properties (to distinguish between "a value of None" and null)
|
||||
cleaned = self._remove_null_properties(cache)
|
||||
if "statuses" in cleaned:
|
||||
cleaned["statuses"] = [status.state for status in cleaned["statuses"]]
|
||||
cleaned["dnssecdata"] = self._get_dnssec_data(data_response.extensions)
|
||||
return cleaned
|
||||
|
||||
def _remove_null_properties(self, cache):
|
||||
return {k: v for k, v in cache.items() if v is not ...}
|
||||
|
||||
|
@ -1679,6 +1650,42 @@ class Domain(TimeStampedModel, DomainHelper):
|
|||
dnssec_data = extension
|
||||
return dnssec_data
|
||||
|
||||
def _update_hosts_and_contacts(self, cleaned, fetch_hosts, fetch_contacts):
|
||||
"""Capture and store old hosts and contacts from cache if the don't exist"""
|
||||
old_cache_hosts = self._cache.get("hosts")
|
||||
old_cache_contacts = self._cache.get("contacts")
|
||||
|
||||
if fetch_contacts:
|
||||
cleaned["contacts"] = self._get_contacts(cleaned.get("_contacts", []))
|
||||
if old_cache_hosts is not None:
|
||||
logger.debug("resetting cleaned['hosts'] to old_cache_hosts")
|
||||
cleaned["hosts"] = old_cache_hosts
|
||||
|
||||
if fetch_hosts:
|
||||
cleaned["hosts"] = self._get_hosts(cleaned.get("_hosts", []))
|
||||
if old_cache_contacts is not None:
|
||||
cleaned["contacts"] = old_cache_contacts
|
||||
|
||||
def _update_dates(self, cleaned):
|
||||
"""Update dates (expiration and creation) from cleaned"""
|
||||
requires_save = False
|
||||
|
||||
# if expiration date from registry does not match what is in db,
|
||||
# update the db
|
||||
if "ex_date" in cleaned and cleaned["ex_date"] != self.expiration_date:
|
||||
self.expiration_date = cleaned["ex_date"]
|
||||
requires_save = True
|
||||
|
||||
# if creation_date from registry does not match what is in db,
|
||||
# update the db
|
||||
if "cr_date" in cleaned and cleaned["cr_date"] != self.created_at:
|
||||
self.created_at = cleaned["cr_date"]
|
||||
requires_save = True
|
||||
|
||||
# if either registration date or creation date need updating
|
||||
if requires_save:
|
||||
self.save()
|
||||
|
||||
def _get_contacts(self, contacts):
|
||||
choices = PublicContact.ContactTypeChoices
|
||||
# We expect that all these fields get populated,
|
||||
|
|
Loading…
Add table
Add a link
Reference in a new issue