15493356367

Committed 06 Jun 2025 02:49PM UTC coverage: 80.424%. First build

Build # 15493356367

Build Type

Pull #249

github

Committed by

web-flow

Commit Message

Merge 3a4745b5e into 5adc47cbe

Pull Request Pull Request #249: Convert to Python 3

Run Details

1628 of 2072 branches covered (78.57%)

21 of 42 new or added lines in 13 files covered. (50.0%)

3829 of 4761 relevant lines covered (80.42%)

4.0 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 = None
    help = 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:
        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
NEW 12	class BaseCommand:	×
13	slug = None	×
14	help = 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
NEW 49	print(f"Listing checkpoints (most recent {args.limit}):")	×
50	if args.project:	×
NEW 51	print(f" project: {args.project}")	×
52	if args.commcare_hq != 'prod':	×
NEW 53	print(f" commcare-hq: {args.commcare_hq}")	×
54	if args.query:	×
NEW 55	print(f" query filename: {args.query}")	×
56	if manager.key:	×
NEW 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:	×
137	sub = subparsers.add_parser(	×
138	command_type.slug,
139	help=inspect.cleandoc(command_type.help).splitlines()[0],
140	description=inspect.cleandoc(command_type.help),
141	formatter_class=argparse.RawDescriptionHelpFormatter
142	)
143	command_type.add_arguments(sub)	×
144
145	args = parser.parse_args(argv)	×
146
147	logging.basicConfig(	×
148	level=logging.WARN,
149	format='%(asctime)s %(name)-12s %(levelname)-8s %(message)s'
150	)
151
152	sys.exit(main_with_args(args))	×
153
154
155	def main_with_args(args):	×
156	command = [c for c in COMMANDS if c.slug == args.command][0]	×
157	command().run(args)	×
158
159
160	def entry_point():	×
161	main(sys.argv[1:])	×
162
163
164	if __name__ == '__main__':	×
165	entry_point()	×

dimagi / commcare-export / 15493356367

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