akvo / iwsims-demo / #72

Committed 28 Apr 2025 03:28AM UTC coverage: 86.134% (+1.1%) from 85.024%

Build # #72

Build Type

push

coveralls-python

Committed by

web-flow

Commit Message

Merge pull request #20 from akvo/feature/19-eng-1231-dynamic-level-approval

Feature/19 eng 1231 dynamic level approval

Run Details

2646 of 3188 branches covered (83.0%)

Branch coverage included in aggregate %.

5995 of 6844 relevant lines covered (87.59%)

0.88 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

91.46

backend/api/v1/v1_profile/management/commands/administration_seeder.py

import pandas as pd
from django.core.management import BaseCommand
from api.v1.v1_profile.models import Levels, Administration
from api.v1.v1_profile.constants import (
    DEFAULT_ADMINISTRATION_DATA,
    DEFAULT_ADMINISTRATION_LEVELS,
    DEFAULT_SOURCE_FILE,
)


def seed_levels(geo_config: list = []):
    for geo in geo_config:
        level = Levels(id=geo["id"], name=geo["alias"], level=geo["level"])
        level.save()


def seed_administration(row: dict, geo_config: list = []):
    for geo in geo_config:
        col_level = f"{geo['level']}_{geo['alias']}"
        parent = None
        if geo["level"] > 0:
            code_parent = row.get(f"{geo['level'] - 1}_code")
            parent = Administration.objects.filter(
                code=code_parent, level__level=geo["level"] - 1
            ).first()
        # Get the level from the geo_config
        level = Levels.objects.filter(level=geo["level"]).first()
        # Get the code from the row
        code = row.get(f"{geo['level']}_code")
        # Get the name from the row
        name = row[col_level]

        Administration.objects.update_or_create(
            name=name,
            defaults={
                "level": level,
                "code": code,
                "parent": parent,
            },
        )


def seed_administration_test(
    rows: list = DEFAULT_ADMINISTRATION_DATA,
    geo_config: list = DEFAULT_ADMINISTRATION_LEVELS,
):
    seed_levels(geo_config=geo_config)
    for row in rows:
        seed_administration(row=row, geo_config=geo_config)


def seed_administration_prod(
    source_file: str = DEFAULT_SOURCE_FILE,
):
    Levels.objects.all().delete()

    df = pd.read_csv(source_file)
    header_columns = df.columns.tolist()
    geo_config = []
    for column in header_columns:
        level = int(column.split("_")[0])
        alias = column.split("_")[1]
        if alias.lower() == "code":
            # Skip the code column
            continue
        geo_config.append(
            {
                "id": level + 1,
                "level": level,
                "name": f"NAME_{level}",
                "alias": alias,
            }
        )
    geo_config = sorted(geo_config, key=lambda x: x["level"])

    seed_levels(geo_config=geo_config)
    df = pd.read_csv(source_file)
    df = df.drop_duplicates()
    df = df.reset_index(drop=True)
    for _, row in df.iterrows():
        seed_administration(row=row, geo_config=geo_config)


class Command(BaseCommand):
    def add_arguments(self, parser):
        parser.add_argument(
            "-t", "--test", nargs="?", const=1, default=False, type=int
        )
        parser.add_argument(
            "-c", "--clean", nargs="?", const=1, default=False, type=int
        )
        parser.add_argument(
            "-s",
            "--source",
            nargs="?",
            const=1,
            default=DEFAULT_SOURCE_FILE,
            type=str,
        )

    def handle(self, *args, **options):
        test = options.get("test")
        clean = options.get("clean")
        source_file = options.get("source")
        if clean:
            Administration.objects.all().delete()
            self.stdout.write("-- Administration Cleared")
        if test:
            seed_administration_test()
        if not test:
            seed_administration_prod(
                source_file=source_file,
            )
            self.stdout.write("-- FINISH")

1	import pandas as pd	1✔
2	from django.core.management import BaseCommand	1✔
3	from api.v1.v1_profile.models import Levels, Administration	1✔
4	from api.v1.v1_profile.constants import (	1✔
5	DEFAULT_ADMINISTRATION_DATA,
6	DEFAULT_ADMINISTRATION_LEVELS,
7	DEFAULT_SOURCE_FILE,
8	)
9
10
11	def seed_levels(geo_config: list = []):	1✔
12	for geo in geo_config:	1✔
13	level = Levels(id=geo["id"], name=geo["alias"], level=geo["level"])	1✔
14	level.save()	1✔
15
16
17	def seed_administration(row: dict, geo_config: list = []):	1✔
18	for geo in geo_config:	1✔
19	col_level = f"{geo['level']}_{geo['alias']}"	1✔
20	parent = None	1✔
21	if geo["level"] > 0:	1✔
22	code_parent = row.get(f"{geo['level'] - 1}_code")	1✔
23	parent = Administration.objects.filter(	1✔
24	code=code_parent, level__level=geo["level"] - 1
25	).first()
26	# Get the level from the geo_config
27	level = Levels.objects.filter(level=geo["level"]).first()	1✔
28	# Get the code from the row
29	code = row.get(f"{geo['level']}_code")	1✔
30	# Get the name from the row
31	name = row[col_level]	1✔
32
33	Administration.objects.update_or_create(	1✔
34	name=name,
35	defaults={
36	"level": level,
37	"code": code,
38	"parent": parent,
39	},
40	)
41
42
43	def seed_administration_test(	1✔
44	rows: list = DEFAULT_ADMINISTRATION_DATA,
45	geo_config: list = DEFAULT_ADMINISTRATION_LEVELS,
46	):
47	seed_levels(geo_config=geo_config)	1✔
48	for row in rows:	1✔
49	seed_administration(row=row, geo_config=geo_config)	1✔
50
51
52	def seed_administration_prod(	1✔
53	source_file: str = DEFAULT_SOURCE_FILE,
54	):
55	Levels.objects.all().delete()	1✔
56
57	df = pd.read_csv(source_file)	1✔
58	header_columns = df.columns.tolist()	1✔
59	geo_config = []	1✔
60	for column in header_columns:	1✔
61	level = int(column.split("_")[0])	1✔
62	alias = column.split("_")[1]	1✔
63	if alias.lower() == "code":	1✔
64	# Skip the code column
65	continue	1✔
66	geo_config.append(	1✔
67	{
68	"id": level + 1,
69	"level": level,
70	"name": f"NAME_{level}",
71	"alias": alias,
72	}
73	)
74	geo_config = sorted(geo_config, key=lambda x: x["level"])	1✔
75
76	seed_levels(geo_config=geo_config)	1✔
77	df = pd.read_csv(source_file)	1✔
78	df = df.drop_duplicates()	1✔
79	df = df.reset_index(drop=True)	1✔
80	for _, row in df.iterrows():	1✔
81	seed_administration(row=row, geo_config=geo_config)	1✔
82
83
84	class Command(BaseCommand):	1✔
85	def add_arguments(self, parser):	1✔
86	parser.add_argument(	1✔
87	"-t", "--test", nargs="?", const=1, default=False, type=int
88	)
89	parser.add_argument(	1✔
90	"-c", "--clean", nargs="?", const=1, default=False, type=int
91	)
92	parser.add_argument(	1✔
93	"-s",
94	"--source",
95	nargs="?",
96	const=1,
97	default=DEFAULT_SOURCE_FILE,
98	type=str,
99	)
100
101	def handle(self, args, *options):	1✔
102	test = options.get("test")	1✔
103	clean = options.get("clean")	1✔
104	source_file = options.get("source")	1✔
105	if clean:	1!
106	Administration.objects.all().delete()	×
107	self.stdout.write("-- Administration Cleared")	×
108	if test:	1!
109	seed_administration_test()	1✔
110	if not test:	1!
111	seed_administration_prod(	×
112	source_file=source_file,
113	)
114	self.stdout.write("-- FINISH")	×

akvo / iwsims-demo / #72

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous