Avoid deeply nested control flow statements.

Open

                        if value:
                            population_data.append(
                                PopulationUrbanRural(country=country, year=year, urban_rural=urban_rural, value=value)
                            )

Avoid deeply nested control flow statements.

Open

                                if user.exists():
                                    user.update(**cleaned_row)
                                    company = user.first().company
                                    company.mobile_number = cleaned_row["mobile_number"]
                                    company.save()

Avoid deeply nested control flow statements.

Open

                        if country_iso3 and uk_or_world:
                            written = written + 1
                            report = ComtradeReport(
                                country_iso3=country_iso3,
                                year=row.get('Year'),

Function process_view has a Cognitive Complexity of 8 (exceeds 5 allowed). Consider refactoring.

Open

    def process_view(self, request, view_func, view_args, view_kwarg):
        if request.user is not None:
            if request.resolver_match.namespace == 'admin' or request.path_info.startswith('/admin/login'):
                if not request.user.is_staff and request.user.is_authenticated:
                    return HttpResponse(self.SSO_UNAUTHORISED_ACCESS_MESSAGE, status=401)

• Read up
  Read up

Function do_import_markets has a Cognitive Complexity of 8 (exceeds 5 allowed). Consider refactoring.

Open

    def do_import_markets(self, *args, **options):
        try:
            with open(self.DEFAULT_FILENAME) as f:
                data = json.load(f)
        except Exception as e:

• Read up
  Read up

Avoid deeply nested control flow statements.

Open

                                if investment_company.exists():
                                    # This will update the object
                                    address_added = self.fetch_companies_house_data(switch_envs)
                                    adjust_types = self.adjust_field_types(address_added)
                                    investment_company.update(**adjust_types)

Function adjust_field_types has a Cognitive Complexity of 8 (exceeds 5 allowed). Consider refactoring.

Open

    def adjust_field_types(self, row: dict) -> dict:
        json_fields = [
            "expertise_products_services",
        ]
        lists = ["expertise_industries", "expertise_regions", "expertise_languages"]

• Read up
  Read up

Avoid deeply nested control flow statements.

Open

                        for k2, v2 in v.items():
                            if not field_value.get(k):
                                # First time this key is being set, default to empty dict so we don't get index error
                                field_value[k] = {}
                            field_value[k][k2] = v2

Avoid deeply nested control flow statements.

Open

                                if investment_company.exists():
                                    # This will update the object
                                    address_added = self.fetch_companies_house_data(cleaned_row)
                                    adjust_types = self.adjust_field_types(address_added)
                                    investment_company.update(**adjust_types)

Avoid deeply nested control flow statements.

Open

                        if partner_iso3 == 'WLD' and flow == 'Import':
                            uk_or_world = partner_iso3
                            country_iso3 = reporter_iso3
                        if country_iso3 and uk_or_world:

Function clean_csv_file has a Cognitive Complexity of 8 (exceeds 5 allowed). Consider refactoring.

Open

    def clean_csv_file(self):
        self.update_errors = []
        self.updated_companies = []

        csv_file = io.TextIOWrapper(self.cleaned_data['csv_file'].file, encoding='utf-8')

• Read up
  Read up

Function should_ingestion_run has a Cognitive Complexity of 8 (exceeds 5 allowed). Consider refactoring.

Open

    def should_ingestion_run(self, view_name, table_name):
        dataflow_metadata = self.get_dataflow_metadata(table_name)
        swapped_date = dataflow_metadata.loc[:, 'dataflow_swapped_tables_utc'][0].to_pydatetime().date()
        great_metadata = self.get_view_metadata(view_name)
        great_metadata_date = None

• Read up
  Read up

Avoid deeply nested control flow statements.

Open

                        if reporter_iso3 == 'GBR' and flow == 'Export':
                            uk_or_world = reporter_iso3
                            country_iso3 = partner_iso3
                        if partner_iso3 == 'WLD' and flow == 'Import':

Function before_import has 5 arguments (exceeds 4 allowed). Consider refactoring.

Open

    def before_import(self, dataset, using_transactions, dry_run, **kwargs):

Function send_account_ownership_transfer_notification has 5 arguments (exceeds 4 allowed). Consider refactoring.

Open

def send_account_ownership_transfer_notification(sender, instance, created, *args, **kwargs):

Function get_multiple_serialized_instance_from_model has 5 arguments (exceeds 4 allowed). Consider refactoring.

Open

def get_multiple_serialized_instance_from_model(model_class, serializer_class, filter_args, section_key, latest_only):

Function send_new_invite_collaboration_notification has 5 arguments (exceeds 4 allowed). Consider refactoring.

Open

def send_new_invite_collaboration_notification(sender, instance, created, *args, **kwargs):

Function send_account_collaborator_notification has 5 arguments (exceeds 4 allowed). Consider refactoring.

Open

def send_account_collaborator_notification(sender, instance, created, *args, **kwargs):

Function bucket_by_country has a Cognitive Complexity of 7 (exceeds 5 allowed). Consider refactoring.

Open

    def bucket_by_country(self, filters, barriers_data):
        bucked_data = {k: {'barriers': []} for (k) in filters.get('locations').keys()}
        for iso_2, name in filters.get('locations', {}).items():
            for barrier in barriers_data:
                if barrier['country']['name'] == name:

• Read up
  Read up

Function handle has a Cognitive Complexity of 7 (exceeds 5 allowed). Consider refactoring.

Open

    def handle(self, *args, **options):
        response = requests.get(
            'https://raw.githubusercontent.com/iancoleman/cia_world_factbook_api/master/data/factbook.json'
        )
        data = response.json()

• Read up
  Read up