zydronium/manage.get.gov
1
0
Fork 0
mirror of https://github.com/cisagov/manage.get.gov.git synced 2025-07-23 19:20:47 +02:00
Code Issues Projects Releases Packages Wiki Activity

Refactored handle() to be less complex, satisfied linter

Browse source 
Signed-off-by: CocoByte <nicolle.leclair@gmail.com>
This commit is contained in:
CocoByte 2023-10-11 16:13:14 -06:00
parent b8a5718113
commit 775ed5f9d2
No known key found for this signature in database
GPG key ID: BBFAA2526384C97F
1 changed files with 105 additions and 88 deletions
Download patch file Download diff file Expand all files Collapse all files

99
src/registrar/management/commands/transfer_transition_domains_to_domains.py
View file

@ -51,29 +51,37 @@ class Command(BaseCommand):
): ):
"""Prints additional terminal statements to indicate if --debug """Prints additional terminal statements to indicate if --debug
or --limitParse are in use""" or --limitParse are in use"""
if debug_on: self.print_debug(
logger.info( debug_on,
f"""{termColors.OKCYAN} f"""{termColors.OKCYAN}
----------DEBUG MODE ON---------- ----------DEBUG MODE ON----------
Detailed print statements activated. Detailed print statements activated.
{termColors.ENDC} {termColors.ENDC}
""" """,
) )
if debug_max_entries_to_parse > 0: self.print_debug(
logger.info( debug_max_entries_to_parse > 0,
f"""{termColors.OKCYAN} f"""{termColors.OKCYAN}
----------LIMITER ON---------- ----------LIMITER ON----------
Parsing of entries will be limited to Parsing of entries will be limited to
{debug_max_entries_to_parse} lines per file.") {debug_max_entries_to_parse} lines per file.")
Detailed print statements activated. Detailed print statements activated.
{termColors.ENDC} {termColors.ENDC}
""" """,
) )
def update_domain_status(self, def print_debug(self, print_condition: bool, print_statement: str):
transition_domain:TransitionDomain, """This function reduces complexity of debug statements
target_domain:Domain, in other functions.
debug_on:bool) -> bool: It uses the logger to write the given print_statement to the
terminal if print_condition is TRUE"""
# DEBUG:
if print_condition:
logger.info(print_statement)
def update_domain_status(
self, transition_domain: TransitionDomain, target_domain: Domain, debug_on: bool
) -> bool:
"""Given a transition domain that matches an existing domain, """Given a transition domain that matches an existing domain,
updates the existing domain object with that status of updates the existing domain object with that status of
the transition domain. the transition domain.
@ -83,31 +91,33 @@ class Command(BaseCommand):
transition_domain_status = transition_domain.status transition_domain_status = transition_domain.status
existing_status = target_domain.state existing_status = target_domain.state
if transition_domain_status != existing_status: if transition_domain_status != existing_status:
if ( if transition_domain_status == TransitionDomain.StatusChoices.ON_HOLD:
transition_domain_status
== TransitionDomain.StatusChoices.ON_HOLD
):
target_domain.place_client_hold(ignoreEPP=True) target_domain.place_client_hold(ignoreEPP=True)
else: else:
target_domain.revert_client_hold(ignoreEPP=True) target_domain.revert_client_hold(ignoreEPP=True)
target_domain.save() target_domain.save()
# DEBUG: # DEBUG:
if debug_on: self.print_debug(
logger.info( debug_on,
f"""{termColors.YELLOW} f"""{termColors.YELLOW}
>> Updated {target_domain.name} state from >> Updated {target_domain.name} state from
'{existing_status}' to '{target_domain.state}' '{existing_status}' to '{target_domain.state}'
(no domain invitation entry added) (no domain invitation entry added)
{termColors.ENDC}""" {termColors.ENDC}""",
) )
return True
return False
def print_summary_of_findings(self, def print_summary_of_findings(
self,
domains_to_create, domains_to_create,
updated_domain_entries, updated_domain_entries,
domain_invitations_to_create, domain_invitations_to_create,
skipped_domain_entries, skipped_domain_entries,
skipped_domain_invitations, skipped_domain_invitations,
debug_on): debug_on,
):
"""Prints to terminal a summary of findings from """Prints to terminal a summary of findings from
transferring transition domains to domains""" transferring transition domains to domains"""
@ -119,7 +129,7 @@ class Command(BaseCommand):
f"""{termColors.OKGREEN} f"""{termColors.OKGREEN}
============= FINISHED =============== ============= FINISHED ===============
Created {total_new_entries} transition domain entries, Created {total_new_entries} transition domain entries,
updated {total_updated_domain_entries} transition domain entries Updated {total_updated_domain_entries} transition domain entries
Created {total_domain_invitation_entries} domain invitation entries Created {total_domain_invitation_entries} domain invitation entries
(NOTE: no invitations are SENT in this script) (NOTE: no invitations are SENT in this script)
@ -144,8 +154,8 @@ class Command(BaseCommand):
) )
# DEBUG: # DEBUG:
if debug_on: self.print_debug(
logger.info( debug_on,
f"""{termColors.YELLOW} f"""{termColors.YELLOW}
Created Domains: Created Domains:
@ -155,7 +165,7 @@ class Command(BaseCommand):
{updated_domain_entries} {updated_domain_entries}
{termColors.ENDC} {termColors.ENDC}
""" """,
) )
def handle( def handle(
@ -203,28 +213,31 @@ class Command(BaseCommand):
# Check for existing domain entry # Check for existing domain entry
try: try:
# DEBUG: # DEBUG:
if debug_on: self.print_debug(
logger.info( debug_on,
f"""{termColors.OKCYAN} f"""{termColors.OKCYAN}
Processing Transition Domain: {transition_domain_name}, {transition_domain_status}, {transition_domain_email} Processing Transition Domain: {transition_domain_name}, {transition_domain_status}, {transition_domain_email}
{termColors.ENDC}""" # noqa {termColors.ENDC}""", # noqa
) )
# get the existing domain # get the existing domain
target_domain = Domain.objects.get(name=transition_domain_name) target_domain = Domain.objects.get(name=transition_domain_name)
# DEBUG: # DEBUG:
if debug_on: self.print_debug(
logger.info( debug_on,
f"""{termColors.YELLOW} f"""{termColors.YELLOW}
> Found existing domain entry for: {transition_domain_name}, {target_domain.state} > Found existing domain entry for: {transition_domain_name}, {target_domain.state}
{termColors.ENDC}""" # noqa {termColors.ENDC}""", # noqa
) )
# for existing entry, update the status to # for existing entry, update the status to
# the transition domain status # the transition domain status
update_made = self.update_domain_status(transition_domain, target_domain, debug_on) update_made = self.update_domain_status(
transition_domain, target_domain, debug_on
)
if update_made: if update_made:
updated_domain_entries.append(transition_domain.name) updated_domain_entries.append(transition_domain.domain_name)
except Domain.DoesNotExist: except Domain.DoesNotExist:
already_in_to_create = next( already_in_to_create = next(
@ -232,14 +245,13 @@ class Command(BaseCommand):
None, None,
) )
if already_in_to_create: if already_in_to_create:
# DEBUG: self.print_debug(
if debug_on: debug_on,
logger.info(
f"""{termColors.YELLOW} f"""{termColors.YELLOW}
Duplicate Detected: {transition_domain_name}. Duplicate Detected: {transition_domain_name}.
Cannot add duplicate entry for another username. Cannot add duplicate entry for another username.
Violates Unique Key constraint. Violates Unique Key constraint.
{termColors.ENDC}""" {termColors.ENDC}""",
) )
else: else:
# no matching entry, make one # no matching entry, make one
@ -261,9 +273,9 @@ class Command(BaseCommand):
skipped_domain_invitations.append(transition_domain_name) skipped_domain_invitations.append(transition_domain_name)
# DEBUG: # DEBUG:
if debug_on: self.print_debug(
logger.info( debug_on,
f"{termColors.OKCYAN} Adding domain AND domain invitation: {new_entry} {termColors.ENDC}" # noqa f"{termColors.OKCYAN} Adding domain AND domain invitation: {new_entry} {termColors.ENDC}", # noqa
) )
except Domain.MultipleObjectsReturned: except Domain.MultipleObjectsReturned:
logger.warning( logger.warning(
@ -286,7 +298,10 @@ class Command(BaseCommand):
) )
# Check parse limit # Check parse limit
if debug_max_entries_to_parse > 0 and total_rows_parsed >= debug_max_entries_to_parse: if (
debug_max_entries_to_parse > 0
and total_rows_parsed >= debug_max_entries_to_parse
):
logger.info( logger.info(
f"""{termColors.YELLOW} f"""{termColors.YELLOW}
----PARSE LIMIT REACHED. HALTING PARSER.---- ----PARSE LIMIT REACHED. HALTING PARSER.----
@ -298,9 +313,11 @@ class Command(BaseCommand):
Domain.objects.bulk_create(domains_to_create) Domain.objects.bulk_create(domains_to_create)
DomainInvitation.objects.bulk_create(domain_invitations_to_create) DomainInvitation.objects.bulk_create(domain_invitations_to_create)
self.print_summary_of_findings(domains_to_create, self.print_summary_of_findings(
domains_to_create,
updated_domain_entries, updated_domain_entries,
domain_invitations_to_create, domain_invitations_to_create,
skipped_domain_entries, skipped_domain_entries,
skipped_domain_invitations, skipped_domain_invitations,
debug_on) debug_on,
)