20852935486

Committed 09 Jan 2026 01:09PM UTC coverage: 81.864%. First build

Build # 20852935486

Build Type

Pull #260

github

Committed by

web-flow

Commit Message

Merge 3e93245a4 into 85d717551

Pull Request Pull Request #260: Check all type annotations. Resolve errors

Run Details

298 of 352 branches covered (84.66%)

28 of 34 new or added lines in 11 files covered. (82.35%)

3918 of 4786 relevant lines covered (81.86%)

4.09 hits per line

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

0.0

/commcare_export/utils_cli.py

import argparse
import inspect
import logging
import sys

from commcare_export.cli import CLI_ARGS
from commcare_export.utils import confirm, get_checkpoint_manager, print_runs

EXIT_STATUS_ERROR = 1


class BaseCommand:
    slug: str | None = None
    help: str | None = None

    @classmethod
    def add_arguments(cls, parser):
        raise NotImplementedError

    def run(self, args):
        raise NotImplementedError


class ListHistoryCommand(BaseCommand):
    slug = 'history'
    help = """List export history. History will be filtered by arguments provided.

    This command only applies when exporting to a SQL database. The command
    lists the checkpoints that have been created by the command.
    """

    @classmethod
    def add_arguments(cls, parser):
        parser.add_argument(
            '--limit',
            default=10,
            help="Limit the number of export runs to display"
        )
        parser.add_argument('--output', required=True, help='SQL Database URL')
        shared_args = {'project', 'query', 'checkpoint_key', 'commcare_hq'}
        for arg in CLI_ARGS:
            if arg.name in shared_args:
                arg.add_to_parser(parser)

    def run(self, args):
        manager = get_checkpoint_manager(args, require_query=False)
        manager.create_checkpoint_table()

        print(f"Listing checkpoints (most recent {args.limit}):")
        if args.project:
            print(f"    project:        {args.project}")
        if args.commcare_hq != 'prod':
            print(f"    commcare-hq:    {args.commcare_hq}")
        if args.query:
            print(f"    query filename: {args.query}")
        if manager.key:
            print(f"    key:            {manager.key}")

        runs = manager.list_checkpoints(args.limit)
        print_runs(runs)


class SetKeyCommand(BaseCommand):
    slug = 'set-checkpoint-key'
    help = """Set the key for a particular checkpoint.

    This command is used to migrate an non-keyed checkpoint to a keyed
    checkpoint.

    This is useful if you already have a populated export database and do
    not wish to trigger rebuilds after editing the query file.

    For example, you've been running the export tool with query file A.xlsx
    and have a fully populated database. Now you need to add an extra column
    to the table but only want to populate it with new data.

    What you need to do is update your current checkpoint with a key that
    you can then use when running the command from now on.

        $ commcare-export-utils set-key --project X --query A.xlsx \\
            --output [SQL URL] --checkpoint-key my-key

    Now when you run the export tool in future you can use this key:

        $ commcare-export --project X --query A.xlsx --output [SQL URL] \\
            --checkpoint-key my-key ...

    """

    @classmethod
    def add_arguments(cls, parser):
        parser.add_argument('--output', required=True, help='SQL Database URL')
        shared_args = {'project', 'query', 'checkpoint_key'}
        for arg in CLI_ARGS:
            if arg.name in shared_args:
                arg.add_to_parser(parser, required=True)
            elif arg.name == 'commcare_hq':
                arg.add_to_parser(parser)

    def run(self, args):
        key = args.checkpoint_key
        manager = get_checkpoint_manager(args)
        manager.create_checkpoint_table()
        run_with_key = manager.list_checkpoints(limit=1)

        if run_with_key:
            print("A checkpoint with that key already exists.")
            return

        manager.key = None
        runs_no_key = manager.get_latest_checkpoints()

        if not runs_no_key:
            print(args)
            print("No checkpoint found with args matching those provided.")
            return

        print_runs(runs_no_key)
        if confirm(
            f"Do you want to set the key for this checkpoint to '{key}'"
        ):
            for checkpoint in runs_no_key:
                checkpoint.key = key
                manager.update_checkpoint(checkpoint)

        print("\nUpdated checkpoint:")
        print_runs(runs_no_key)


COMMANDS = [ListHistoryCommand, SetKeyCommand]


def main(argv):
    parser = argparse.ArgumentParser('commcare-export-utils')
    subparsers = parser.add_subparsers(dest='command')
    for command_type in COMMANDS:
        assert command_type.slug is not None
        assert command_type.help is not None
        sub = subparsers.add_parser(
            command_type.slug,
            help=inspect.cleandoc(command_type.help).splitlines()[0],
            description=inspect.cleandoc(command_type.help),
            formatter_class=argparse.RawDescriptionHelpFormatter
        )
        command_type.add_arguments(sub)

    args = parser.parse_args(argv)

    logging.basicConfig(
        level=logging.WARN,
        format='%(asctime)s %(name)-12s %(levelname)-8s %(message)s'
    )

    sys.exit(main_with_args(args))


def main_with_args(args):
    command = [c for c in COMMANDS if c.slug == args.command][0]
    command().run(args)


def entry_point():
    main(sys.argv[1:])


if __name__ == '__main__':
    entry_point()

1	import argparse	×
2	import inspect	×
3	import logging	×
4	import sys	×
5
6	from commcare_export.cli import CLI_ARGS	×
7	from commcare_export.utils import confirm, get_checkpoint_manager, print_runs	×
8
9	EXIT_STATUS_ERROR = 1	×
10
11
12	class BaseCommand:	×
NEW 13	slug: str \| None = None	×
NEW 14	help: str \| None = None	×
15
16	@classmethod	×
17	def add_arguments(cls, parser):	×
18	raise NotImplementedError	×
19
20	def run(self, args):	×
21	raise NotImplementedError	×
22
23
24	class ListHistoryCommand(BaseCommand):	×
25	slug = 'history'	×
26	help = """List export history. History will be filtered by arguments provided.	×
27
28	This command only applies when exporting to a SQL database. The command
29	lists the checkpoints that have been created by the command.
30	"""
31
32	@classmethod	×
33	def add_arguments(cls, parser):	×
34	parser.add_argument(	×
35	'--limit',
36	default=10,
37	help="Limit the number of export runs to display"
38	)
39	parser.add_argument('--output', required=True, help='SQL Database URL')	×
40	shared_args = {'project', 'query', 'checkpoint_key', 'commcare_hq'}	×
41	for arg in CLI_ARGS:	×
42	if arg.name in shared_args:	×
43	arg.add_to_parser(parser)	×
44
45	def run(self, args):	×
46	manager = get_checkpoint_manager(args, require_query=False)	×
47	manager.create_checkpoint_table()	×
48
49	print(f"Listing checkpoints (most recent {args.limit}):")	×
50	if args.project:	×
51	print(f" project: {args.project}")	×
52	if args.commcare_hq != 'prod':	×
53	print(f" commcare-hq: {args.commcare_hq}")	×
54	if args.query:	×
55	print(f" query filename: {args.query}")	×
56	if manager.key:	×
57	print(f" key: {manager.key}")	×
58
59	runs = manager.list_checkpoints(args.limit)	×
60	print_runs(runs)	×
61
62
63	class SetKeyCommand(BaseCommand):	×
64	slug = 'set-checkpoint-key'	×
65	help = """Set the key for a particular checkpoint.	×
66
67	This command is used to migrate an non-keyed checkpoint to a keyed
68	checkpoint.
69
70	This is useful if you already have a populated export database and do
71	not wish to trigger rebuilds after editing the query file.
72
73	For example, you've been running the export tool with query file A.xlsx
74	and have a fully populated database. Now you need to add an extra column
75	to the table but only want to populate it with new data.
76
77	What you need to do is update your current checkpoint with a key that
78	you can then use when running the command from now on.
79
80	$ commcare-export-utils set-key --project X --query A.xlsx \\
81	--output [SQL URL] --checkpoint-key my-key
82
83	Now when you run the export tool in future you can use this key:
84
85	$ commcare-export --project X --query A.xlsx --output [SQL URL] \\
86	--checkpoint-key my-key ...
87
88	"""
89
90	@classmethod	×
91	def add_arguments(cls, parser):	×
92	parser.add_argument('--output', required=True, help='SQL Database URL')	×
93	shared_args = {'project', 'query', 'checkpoint_key'}	×
94	for arg in CLI_ARGS:	×
95	if arg.name in shared_args:	×
96	arg.add_to_parser(parser, required=True)	×
97	elif arg.name == 'commcare_hq':	×
98	arg.add_to_parser(parser)	×
99
100	def run(self, args):	×
101	key = args.checkpoint_key	×
102	manager = get_checkpoint_manager(args)	×
103	manager.create_checkpoint_table()	×
104	run_with_key = manager.list_checkpoints(limit=1)	×
105
106	if run_with_key:	×
107	print("A checkpoint with that key already exists.")	×
108	return	×
109
110	manager.key = None	×
111	runs_no_key = manager.get_latest_checkpoints()	×
112
113	if not runs_no_key:	×
114	print(args)	×
115	print("No checkpoint found with args matching those provided.")	×
116	return	×
117
118	print_runs(runs_no_key)	×
119	if confirm(	×
120	f"Do you want to set the key for this checkpoint to '{key}'"
121	):
122	for checkpoint in runs_no_key:	×
123	checkpoint.key = key	×
124	manager.update_checkpoint(checkpoint)	×
125
126	print("\nUpdated checkpoint:")	×
127	print_runs(runs_no_key)	×
128
129
130	COMMANDS = [ListHistoryCommand, SetKeyCommand]	×
131
132
133	def main(argv):	×
134	parser = argparse.ArgumentParser('commcare-export-utils')	×
135	subparsers = parser.add_subparsers(dest='command')	×
136	for command_type in COMMANDS:	×
NEW 137	assert command_type.slug is not None	×
NEW 138	assert command_type.help is not None	×
139	sub = subparsers.add_parser(	×
140	command_type.slug,
141	help=inspect.cleandoc(command_type.help).splitlines()[0],
142	description=inspect.cleandoc(command_type.help),
143	formatter_class=argparse.RawDescriptionHelpFormatter
144	)
145	command_type.add_arguments(sub)	×
146
147	args = parser.parse_args(argv)	×
148
149	logging.basicConfig(	×
150	level=logging.WARN,
151	format='%(asctime)s %(name)-12s %(levelname)-8s %(message)s'
152	)
153
154	sys.exit(main_with_args(args))	×
155
156
157	def main_with_args(args):	×
158	command = [c for c in COMMANDS if c.slug == args.command][0]	×
159	command().run(args)	×
160
161
162	def entry_point():	×
163	main(sys.argv[1:])	×
164
165
166	if __name__ == '__main__':	×
167	entry_point()	×

dimagi / commcare-export / 20852935486

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