8559166587

Committed 04 Apr 2024 05:58PM UTC coverage: 80.158%. First build

Build # 8559166587

Build Type

Pull #239

github

Committed by

web-flow

Commit Message

Merge a4fcf3fb1 into f48e7a9a8

Pull Request Pull Request #239: Improve DET errors

Run Details

1759 of 2244 branches covered (78.39%)

11 of 16 new or added lines in 1 file covered. (68.75%)

1931 of 2409 relevant lines covered (80.16%)

4.01 hits per line

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

53.45

/commcare_export/cli.py

import argparse
import getpass
import io
import json
import os.path
import sys
import logging
import dateutil.parser
import requests
import sqlalchemy

from commcare_export import builtin_queries, excel_query, writers
from commcare_export.checkpoint import CheckpointManagerProvider
from commcare_export.commcare_hq_client import (
    LATEST_KNOWN_VERSION,
    CommCareHqClient,
    ResourceRepeatException,
)
from commcare_export.commcare_minilinq import CommCareHqEnv
from commcare_export.env import BuiltInEnv, EmitterEnv, JsonPathEnv
from commcare_export.exceptions import (
    DataExportException,
    MissingQueryFileException,
)
from commcare_export.location_info_provider import LocationInfoProvider
from commcare_export.minilinq import List, MiniLinq
from commcare_export.misc import default_to_json
from commcare_export.repeatable_iterator import RepeatableIterator
from commcare_export.utils import get_checkpoint_manager
from commcare_export.version import __version__
from commcare_export import get_logger, get_error_logger

EXIT_STATUS_ERROR = 1
logger = get_logger(__file__)

commcare_hq_aliases = {
    'local': 'http://localhost:8000',
    'prod': 'https://www.commcarehq.org'
}


class Argument(object):

    def __init__(self, name, *args, **kwargs):
        self.name = name.replace('-', '_')
        self._args = ['--{}'.format(name)] + list(args)
        self._kwargs = kwargs

    @property
    def default(self):
        return self._kwargs.get('default')

    def add_to_parser(self, parser, **additional_kwargs):
        additional_kwargs.update(self._kwargs)
        parser.add_argument(*self._args, **additional_kwargs)


CLI_ARGS = [
    Argument(
        'version',
        default=False,
        action='store_true',
        help='Print the current version of the commcare-export tool.'
    ),
    Argument('query', required=False, help='JSON or Excel query file'),
    Argument('dump-query', default=False, action='store_true'),
    Argument(
        'commcare-hq',
        default='prod',
        help='Base url for the CommCare HQ instance e.g. '
        'https://www.commcarehq.org'
    ),
    Argument('api-version', default=LATEST_KNOWN_VERSION),
    Argument('project'),
    Argument('username'),
    Argument(
        'password',
        help='Enter password, or if using apikey auth-mode, enter the api key.'
    ),
    Argument(
        'auth-mode',
        default='password',
        choices=['password', 'apikey'],
        help='Use "digest" auth, or "apikey" auth (for two factor enabled '
        'domains).'
    ),
    Argument(
        'since',
        help='Export all data after this date. Format YYYY-MM-DD or '
        'YYYY-MM-DDTHH:mm:SS'
    ),
    Argument(
        'until',
        help='Export all data up until this date. Format YYYY-MM-DD or '
        'YYYY-MM-DDTHH:mm:SS'
    ),
    Argument(
        'start-over',
        default=False,
        action='store_true',
        help='When saving to a SQL database; the default is to pick up '
        'since the last success. This disables that.'
    ),
    Argument('profile'),
    Argument('verbose', default=False, action='store_true'),
    Argument(
        'output-format',
        default='json',
        choices=['json', 'csv', 'xls', 'xlsx', 'sql', 'markdown'],
        help='Output format'
    ),
    Argument(
        'output',
        metavar='PATH',
        default='reports.zip',
        help='Path to output; defaults to `reports.zip`.'
    ),
    Argument(
        'strict-types',
        default=False,
        action='store_true',
        help="When saving to a SQL database don't allow changing column types "
        "once they are created."
    ),
    Argument(
        'missing-value',
        default=None,
        help="Value to use when a field is missing from the form / case."
    ),
    Argument(
        'batch-size',
        default=200,
        help="Number of records to process per batch."
    ),
    Argument(
        'checkpoint-key',
        help="Use this key for all checkpoints instead of the query file MD5 "
        "hash in order to prevent table rebuilds after a query file has "
        "been edited."
    ),
    Argument(
        'users',
        default=False,
        action='store_true',
        help="Export a table containing data about this project's mobile "
        "workers"
    ),
    Argument(
        'locations',
        default=False,
        action='store_true',
        help="Export a table containing data about this project's locations"
    ),
    Argument(
        'with-organization',
        default=False,
        action='store_true',
        help="Export tables containing mobile worker data and location data "
        "and add a commcare_userid field to any exported form or case"
    ),
    Argument(
        'export-root-if-no-subdocument',
        default=False,
        action='store_true',
        help="Use this when you are exporting a nested document e.g. "
        "form.form..case, messaging-event.messages.[*] And you want to "
        "have a record exported even if the nested document does not "
        "exist or is empty.",
    ),
    Argument(
        'no-logfile',
        default=False,
        help="Specify in order to prevent information being logged to the log file and"
             " show all output in the console.",
        action='store_true',
    ),
]


def main(argv):
    parser = argparse.ArgumentParser(
        'commcare-export', 'Output a customized export of CommCareHQ data.'
    )
    for arg in CLI_ARGS:
        arg.add_to_parser(parser)

    args = parser.parse_args(argv)

    if args.output_format and args.output:
        errors = []
        errors.extend(validate_output_filename(args.output_format, args.output))
        if errors:
            raise Exception(f"Could not proceed. Following issues were found: {', '.join(errors)}.")

    if not args.no_logfile:
        exe_dir = os.path.dirname(sys.executable)
        log_file = os.path.join(exe_dir, "commcare_export.log")
        print(f"Printing logs to {log_file}")
        logging.basicConfig(
            filename=log_file,
            format='%(asctime)s %(name)-12s %(levelname)-8s %(message)s',
            filemode='w',
        )
        sys.stderr = get_error_logger()

    if args.verbose:
        logging.basicConfig(
            level=logging.DEBUG,
            format='%(asctime)s %(name)-12s %(levelname)-8s %(message)s'
        )
    else:
        logging.basicConfig(
            level=logging.WARN,
            format='%(asctime)s %(name)-12s %(levelname)-8s %(message)s'
        )

    logging.getLogger('alembic').setLevel(logging.WARN)
    logging.getLogger('backoff').setLevel(logging.FATAL)
    logging.getLogger('urllib3').setLevel(logging.WARN)

    if args.version:
        print('commcare-export version {}'.format(__version__))
        sys.exit(0)

    if not args.project:
        error_msg = "commcare-export: error: argument --project is required"
        # output to log file through sys.stderr
        print(
            error_msg,
            file=sys.stderr
        )
        # Output to console for debugging
        print(error_msg)
        sys.exit(1)

    if args.profile:
        # hotshot is gone in Python 3
        import hotshot
        import hotshot.stats
        profile = hotshot.Profile(args.profile)
        profile.start()

    try:
        print("Running export...")
        try:
            exit_code = main_with_args(args)
            if exit_code > 0:
                print("Error occurred! See log file for error.")
            sys.exit(exit_code)
        except Exception:
            print("Error occurred! See log file for error.")
            raise
    finally:
        print("Export finished!")
        if args.profile:
            profile.close()
            stats = hotshot.stats.load(args.profile)
            stats.strip_dirs()
            stats.sort_stats('cumulative', 'calls')
            stats.print_stats(100)


def validate_output_filename(output_format, output_filename):
    """
    Validate file extensions for csv, xls and xlsx output formats.
    Ensure extension unless using sql output_format.
    """
    errors = []
    if output_format == 'csv' and not output_filename.endswith('.zip'):
        errors.append("For output format as csv, output file name should have extension zip")
    elif output_format == 'xls' and not output_filename.endswith('.xls'):
        errors.append("For output format as xls, output file name should have extension xls")
    elif output_format == 'xlsx' and not output_filename.endswith('.xlsx'):
        errors.append("For output format as xlsx, output file name should have extension xlsx")
    elif output_format != 'sql' and "." not in output_filename:
        errors.append("Missing extension in output file name")
    return errors


def _get_query(args, writer, column_enforcer=None):
    return _get_query_from_file(
        args.query,
        args.missing_value,
        writer.supports_multi_table_write,
        writer.max_column_length,
        writer.required_columns,
        column_enforcer,
        args.export_root_if_no_subdocument
    )


def _get_query_from_file(
    query_arg,
    missing_value,
    combine_emits,
    max_column_length,
    required_columns,
    column_enforcer,
    value_or_root
):
    if os.path.exists(query_arg):
        if os.path.splitext(query_arg)[1] in ['.xls', '.xlsx']:
            import openpyxl
            workbook = openpyxl.load_workbook(query_arg)
            return excel_query.get_queries_from_excel(
                workbook,
                missing_value,
                combine_emits,
                max_column_length,
                required_columns,
                column_enforcer,
                value_or_root
            )
        else:
            with io.open(query_arg, encoding='utf-8') as fh:
                return MiniLinq.from_jvalue(json.loads(fh.read()))


def get_queries(args, writer, lp, column_enforcer=None):
    query_list = []
    if args.query is not None:
        query = _get_query(args, writer, column_enforcer=column_enforcer)

        if not query:
            raise MissingQueryFileException(args.query)
        query_list.append(query)

    if args.users or args.with_organization:
        # Add user data to query
        query_list.append(builtin_queries.users_query)

    if args.locations or args.with_organization:
        # Add location data to query
        query_list.append(builtin_queries.get_locations_query(lp))

    return List(query_list) if len(query_list) > 1 else query_list[0]


def _get_writer(output_format, output, strict_types):
    if output_format == 'xlsx':
        return writers.Excel2007TableWriter(output)
    elif output_format == 'xls':
        return writers.Excel2003TableWriter(output)
    elif output_format == 'csv':
        if not output.endswith(".zip"):
            print(
                "WARNING: csv output is a zip file, but "
                "will be written to %s" % output
            )
            print(
                "Consider appending .zip to the file name to avoid confusion."
            )
        return writers.CsvTableWriter(output)
    elif output_format == 'json':
        return writers.JValueTableWriter()
    elif output_format == 'markdown':
        return writers.StreamingMarkdownTableWriter(sys.stdout)
    elif output_format == 'sql':
        # Output should be a connection URL. Writer had bizarre issues
        # so we use a full connection instead of passing in a URL or
        # engine.
        if output.startswith('mysql'):
            charset_split = output.split('charset=')
            if len(charset_split) > 1 and charset_split[1] != 'utf8mb4':
                raise Exception(
                    f"The charset '{charset_split[1]}' might cause problems with the export. "
                    f"It is recommended that you use 'utf8mb4' instead."
                )

        return writers.SqlTableWriter(output, strict_types)
    else:
        raise Exception("Unknown output format: {}".format(output_format))


def get_date_params(args):
    since = dateutil.parser.parse(args.since) if args.since else None
    until = dateutil.parser.parse(args.until) if args.until else None
    return since, until


def _get_api_client(args, commcarehq_base_url):
    return CommCareHqClient(
        url=commcarehq_base_url,
        project=args.project,
        username=args.username,
        password=args.password,
        auth_mode=args.auth_mode,
        version=args.api_version
    )


def _get_checkpoint_manager(args):
    if not args.users and not args.locations and not os.path.exists(
        args.query
    ):
        logger.warning(
            "Checkpointing disabled for non builtin, "
            "non file-based query"
        )
    elif args.since or args.until:
        logger.warning(
            "Checkpointing disabled when using '--since' or '--until'"
        )
    else:
        checkpoint_manager = get_checkpoint_manager(args)
        checkpoint_manager.create_checkpoint_table()
        return checkpoint_manager


def force_lazy_result(lazy_result):
    if lazy_result is not None:
        if isinstance(lazy_result, RepeatableIterator):
            list(lazy_result) if lazy_result else lazy_result
        else:
            for nested_result in lazy_result:
                force_lazy_result(nested_result)


def evaluate_query(env, query):
    with env:
        try:
            lazy_result = query.eval(env)
            force_lazy_result(lazy_result)
            return 0
        except requests.exceptions.RequestException as e:
            if e.response and e.response.status_code == 401:
                print(
                    "\nAuthentication failed. Please check your credentials.",
                    file=sys.stderr
                )
                return EXIT_STATUS_ERROR
            else:
                raise
        except ResourceRepeatException as e:
            print('Stopping because the export is stuck')
            print(e.message)
            print('Try increasing --batch-size to overcome the error')
            return EXIT_STATUS_ERROR
        except (
            sqlalchemy.exc.DataError,
            sqlalchemy.exc.InternalError,
            sqlalchemy.exc.ProgrammingError
        ) as e:
            print('Stopping because of database error:\n', e)
            return EXIT_STATUS_ERROR
        except KeyboardInterrupt:
            print('\nExport aborted', file=sys.stderr)
            return EXIT_STATUS_ERROR


def main_with_args(args):
    logger.info("CommCare Export Version {}".format(__version__))
    writer = _get_writer(args.output_format, args.output, args.strict_types)

    if args.query is None and args.users is False and args.locations is False:
        print(
            'At least one the following arguments is required: '
            '--query, --users, --locations',
            file=sys.stderr
        )
        return EXIT_STATUS_ERROR

    if not args.username:
        logger.warn("Username not provided")
        args.username = input('Please provide a username: ')

    if not args.password:
        logger.warn("Password not provided")
        # Windows getpass does not accept unicode
        args.password = getpass.getpass()

    column_enforcer = None
    if args.with_organization:
        column_enforcer = builtin_queries.ColumnEnforcer()

    commcarehq_base_url = commcare_hq_aliases.get(
        args.commcare_hq, args.commcare_hq
    )
    api_client = _get_api_client(args, commcarehq_base_url)
    lp = LocationInfoProvider(api_client, page_size=args.batch_size)
    try:
        query = get_queries(args, writer, lp, column_enforcer)
    except DataExportException as e:
        print(e.message, file=sys.stderr)
        return EXIT_STATUS_ERROR

    if args.dump_query:
        print(json.dumps(query.to_jvalue(), indent=4))
        return

    checkpoint_manager = None
    if writer.support_checkpoints:
        checkpoint_manager = _get_checkpoint_manager(args)

    since, until = get_date_params(args)
    if args.start_over:
        if checkpoint_manager:
            logger.warning(
                'Ignoring all checkpoints and re-fetching all data from '
                'CommCare.'
            )
    elif since:
        logger.debug('Starting from %s', args.since)

    cm = CheckpointManagerProvider(checkpoint_manager, since, args.start_over)
    static_env = {
        'commcarehq_base_url': commcarehq_base_url,
        'get_checkpoint_manager': cm.get_checkpoint_manager,
        'get_location_info': lp.get_location_info,
        'get_location_ancestor': lp.get_location_ancestor
    }
    env = (
        BuiltInEnv(static_env)
        | CommCareHqEnv(api_client, until=until, page_size=args.batch_size)
        | JsonPathEnv({})
        | EmitterEnv(writer)
    )

    exit_status = evaluate_query(env, query)

    if args.output_format == 'json':
        print(
            json.dumps(
                list(writer.tables.values()),
                indent=4,
                default=default_to_json
            )
        )

    return exit_status


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


if __name__ == '__main__':
    entry_point()

1	import argparse	5✔
2	import getpass	5✔
3	import io	5✔
4	import json	5✔
5	import os.path	5✔
6	import sys	5✔
7	import logging	5✔
8	import dateutil.parser	5✔
9	import requests	5✔
10	import sqlalchemy	5✔
11
12	from commcare_export import builtin_queries, excel_query, writers	5✔
13	from commcare_export.checkpoint import CheckpointManagerProvider	5✔
14	from commcare_export.commcare_hq_client import (	5✔
15	LATEST_KNOWN_VERSION,
16	CommCareHqClient,
17	ResourceRepeatException,
18	)
19	from commcare_export.commcare_minilinq import CommCareHqEnv	5✔
20	from commcare_export.env import BuiltInEnv, EmitterEnv, JsonPathEnv	5✔
21	from commcare_export.exceptions import (	5✔
22	DataExportException,
23	MissingQueryFileException,
24	)
25	from commcare_export.location_info_provider import LocationInfoProvider	5✔
26	from commcare_export.minilinq import List, MiniLinq	5✔
27	from commcare_export.misc import default_to_json	5✔
28	from commcare_export.repeatable_iterator import RepeatableIterator	5✔
29	from commcare_export.utils import get_checkpoint_manager	5✔
30	from commcare_export.version import __version__	5✔
31	from commcare_export import get_logger, get_error_logger	5✔
32
33	EXIT_STATUS_ERROR = 1	5✔
34	logger = get_logger(__file__)	5✔
35
36	commcare_hq_aliases = {	5✔
37	'local': 'http://localhost:8000',
38	'prod': 'https://www.commcarehq.org'
39	}
40
41
42	class Argument(object):	5✔
43
44	def __init__(self, name, args, *kwargs):	5✔
45	self.name = name.replace('-', '_')	5✔
46	self._args = ['--{}'.format(name)] + list(args)	5✔
47	self._kwargs = kwargs	5✔
48
49	@property	5✔
50	def default(self):	5✔
51	return self._kwargs.get('default')	5✔
52
53	def add_to_parser(self, parser, **additional_kwargs):	5✔
54	additional_kwargs.update(self._kwargs)	×
55	parser.add_argument(self._args, *additional_kwargs)	×
56
57
58	CLI_ARGS = [	5✔
59	Argument(
60	'version',
61	default=False,
62	action='store_true',
63	help='Print the current version of the commcare-export tool.'
64	),
65	Argument('query', required=False, help='JSON or Excel query file'),
66	Argument('dump-query', default=False, action='store_true'),
67	Argument(
68	'commcare-hq',
69	default='prod',
70	help='Base url for the CommCare HQ instance e.g. '
71	'https://www.commcarehq.org'
72	),
73	Argument('api-version', default=LATEST_KNOWN_VERSION),
74	Argument('project'),
75	Argument('username'),
76	Argument(
77	'password',
78	help='Enter password, or if using apikey auth-mode, enter the api key.'
79	),
80	Argument(
81	'auth-mode',
82	default='password',
83	choices=['password', 'apikey'],
84	help='Use "digest" auth, or "apikey" auth (for two factor enabled '
85	'domains).'
86	),
87	Argument(
88	'since',
89	help='Export all data after this date. Format YYYY-MM-DD or '
90	'YYYY-MM-DDTHH:mm:SS'
91	),
92	Argument(
93	'until',
94	help='Export all data up until this date. Format YYYY-MM-DD or '
95	'YYYY-MM-DDTHH:mm:SS'
96	),
97	Argument(
98	'start-over',
99	default=False,
100	action='store_true',
101	help='When saving to a SQL database; the default is to pick up '
102	'since the last success. This disables that.'
103	),
104	Argument('profile'),
105	Argument('verbose', default=False, action='store_true'),
106	Argument(
107	'output-format',
108	default='json',
109	choices=['json', 'csv', 'xls', 'xlsx', 'sql', 'markdown'],
110	help='Output format'
111	),
112	Argument(
113	'output',
114	metavar='PATH',
115	default='reports.zip',
116	help='Path to output; defaults to `reports.zip`.'
117	),
118	Argument(
119	'strict-types',
120	default=False,
121	action='store_true',
122	help="When saving to a SQL database don't allow changing column types "
123	"once they are created."
124	),
125	Argument(
126	'missing-value',
127	default=None,
128	help="Value to use when a field is missing from the form / case."
129	),
130	Argument(
131	'batch-size',
132	default=200,
133	help="Number of records to process per batch."
134	),
135	Argument(
136	'checkpoint-key',
137	help="Use this key for all checkpoints instead of the query file MD5 "
138	"hash in order to prevent table rebuilds after a query file has "
139	"been edited."
140	),
141	Argument(
142	'users',
143	default=False,
144	action='store_true',
145	help="Export a table containing data about this project's mobile "
146	"workers"
147	),
148	Argument(
149	'locations',
150	default=False,
151	action='store_true',
152	help="Export a table containing data about this project's locations"
153	),
154	Argument(
155	'with-organization',
156	default=False,
157	action='store_true',
158	help="Export tables containing mobile worker data and location data "
159	"and add a commcare_userid field to any exported form or case"
160	),
161	Argument(
162	'export-root-if-no-subdocument',
163	default=False,
164	action='store_true',
165	help="Use this when you are exporting a nested document e.g. "
166	"form.form..case, messaging-event.messages.[*] And you want to "
167	"have a record exported even if the nested document does not "
168	"exist or is empty.",
169	),
170	Argument(
171	'no-logfile',
172	default=False,
173	help="Specify in order to prevent information being logged to the log file and"
174	" show all output in the console.",
175	action='store_true',
176	),
177	]
178
179
180	def main(argv):	5✔
181	parser = argparse.ArgumentParser(	×
182	'commcare-export', 'Output a customized export of CommCareHQ data.'
183	)
184	for arg in CLI_ARGS:	×
185	arg.add_to_parser(parser)	×
186
187	args = parser.parse_args(argv)	×
188
NEW 189	if args.output_format and args.output:	×
NEW 190	errors = []	×
NEW 191	errors.extend(validate_output_filename(args.output_format, args.output))	×
NEW 192	if errors:	×
NEW 193	raise Exception(f"Could not proceed. Following issues were found: {', '.join(errors)}.")	×
194
195	if not args.no_logfile:	×
196	exe_dir = os.path.dirname(sys.executable)	×
197	log_file = os.path.join(exe_dir, "commcare_export.log")	×
198	print(f"Printing logs to {log_file}")	×
199	logging.basicConfig(	×
200	filename=log_file,
201	format='%(asctime)s %(name)-12s %(levelname)-8s %(message)s',
202	filemode='w',
203	)
204	sys.stderr = get_error_logger()	×
205
206	if args.verbose:	×
207	logging.basicConfig(	×
208	level=logging.DEBUG,
209	format='%(asctime)s %(name)-12s %(levelname)-8s %(message)s'
210	)
211	else:
212	logging.basicConfig(	×
213	level=logging.WARN,
214	format='%(asctime)s %(name)-12s %(levelname)-8s %(message)s'
215	)
216
217	logging.getLogger('alembic').setLevel(logging.WARN)	×
218	logging.getLogger('backoff').setLevel(logging.FATAL)	×
219	logging.getLogger('urllib3').setLevel(logging.WARN)	×
220
221	if args.version:	×
222	print('commcare-export version {}'.format(__version__))	×
223	sys.exit(0)	×
224
225	if not args.project:	×
226	error_msg = "commcare-export: error: argument --project is required"	×
227	# output to log file through sys.stderr
228	print(	×
229	error_msg,
230	file=sys.stderr
231	)
232	# Output to console for debugging
233	print(error_msg)	×
234	sys.exit(1)	×
235
236	if args.profile:	×
237	# hotshot is gone in Python 3
238	import hotshot	×
239	import hotshot.stats	×
240	profile = hotshot.Profile(args.profile)	×
241	profile.start()	×
242
243	try:	×
244	print("Running export...")	×
245	try:	×
246	exit_code = main_with_args(args)	×
247	if exit_code > 0:	×
248	print("Error occurred! See log file for error.")	×
249	sys.exit(exit_code)	×
250	except Exception:	×
251	print("Error occurred! See log file for error.")	×
252	raise	×
253	finally:
254	print("Export finished!")	×
255	if args.profile:	×
256	profile.close()	×
257	stats = hotshot.stats.load(args.profile)	×
258	stats.strip_dirs()	×
259	stats.sort_stats('cumulative', 'calls')	×
260	stats.print_stats(100)	×
261
262
263	def validate_output_filename(output_format, output_filename):	5✔
264	"""
265	Validate file extensions for csv, xls and xlsx output formats.
266	Ensure extension unless using sql output_format.
267	"""
268	errors = []	5✔
269	if output_format == 'csv' and not output_filename.endswith('.zip'):	5✔
270	errors.append("For output format as csv, output file name should have extension zip")	5✔
271	elif output_format == 'xls' and not output_filename.endswith('.xls'):	5✔
272	errors.append("For output format as xls, output file name should have extension xls")	5✔
273	elif output_format == 'xlsx' and not output_filename.endswith('.xlsx'):	5✔
274	errors.append("For output format as xlsx, output file name should have extension xlsx")	5✔
275	elif output_format != 'sql' and "." not in output_filename:	5✔
276	errors.append("Missing extension in output file name")	5✔
277	return errors	5✔
278
279
280	def _get_query(args, writer, column_enforcer=None):	5✔
281	return _get_query_from_file(	5✔
282	args.query,
283	args.missing_value,
284	writer.supports_multi_table_write,
285	writer.max_column_length,
286	writer.required_columns,
287	column_enforcer,
288	args.export_root_if_no_subdocument
289	)
290
291
292	def _get_query_from_file(	5✔
293	query_arg,
294	missing_value,
295	combine_emits,
296	max_column_length,
297	required_columns,
298	column_enforcer,
299	value_or_root
300	):
301	if os.path.exists(query_arg):	5✔
302	if os.path.splitext(query_arg)[1] in ['.xls', '.xlsx']:	5✔
303	import openpyxl	5✔
304	workbook = openpyxl.load_workbook(query_arg)	5✔
305	return excel_query.get_queries_from_excel(	5✔
306	workbook,
307	missing_value,
308	combine_emits,
309	max_column_length,
310	required_columns,
311	column_enforcer,
312	value_or_root
313	)
314	else:
315	with io.open(query_arg, encoding='utf-8') as fh:	×
316	return MiniLinq.from_jvalue(json.loads(fh.read()))	×
317
318
319	def get_queries(args, writer, lp, column_enforcer=None):	5✔
320	query_list = []	5✔
321	if args.query is not None:	5✔
322	query = _get_query(args, writer, column_enforcer=column_enforcer)	5✔
323
324	if not query:	5✔
325	raise MissingQueryFileException(args.query)	×
326	query_list.append(query)	5✔
327
328	if args.users or args.with_organization:	5✔
329	# Add user data to query
330	query_list.append(builtin_queries.users_query)	5✔
331
332	if args.locations or args.with_organization:	5✔
333	# Add location data to query
334	query_list.append(builtin_queries.get_locations_query(lp))	5✔
335
336	return List(query_list) if len(query_list) > 1 else query_list[0]	5✔
337
338
339	def _get_writer(output_format, output, strict_types):	5✔
340	if output_format == 'xlsx':	×
341	return writers.Excel2007TableWriter(output)	×
342	elif output_format == 'xls':	×
343	return writers.Excel2003TableWriter(output)	×
344	elif output_format == 'csv':	×
345	if not output.endswith(".zip"):	×
346	print(	×
347	"WARNING: csv output is a zip file, but "
348	"will be written to %s" % output
349	)
350	print(	×
351	"Consider appending .zip to the file name to avoid confusion."
352	)
353	return writers.CsvTableWriter(output)	×
354	elif output_format == 'json':	×
355	return writers.JValueTableWriter()	×
356	elif output_format == 'markdown':	×
357	return writers.StreamingMarkdownTableWriter(sys.stdout)	×
358	elif output_format == 'sql':	×
359	# Output should be a connection URL. Writer had bizarre issues
360	# so we use a full connection instead of passing in a URL or
361	# engine.
362	if output.startswith('mysql'):	×
363	charset_split = output.split('charset=')	×
364	if len(charset_split) > 1 and charset_split[1] != 'utf8mb4':	×
365	raise Exception(	×
366	f"The charset '{charset_split[1]}' might cause problems with the export. "
367	f"It is recommended that you use 'utf8mb4' instead."
368	)
369
370	return writers.SqlTableWriter(output, strict_types)	×
371	else:
372	raise Exception("Unknown output format: {}".format(output_format))	×
373
374
375	def get_date_params(args):	5✔
376	since = dateutil.parser.parse(args.since) if args.since else None	5✔
377	until = dateutil.parser.parse(args.until) if args.until else None	5✔
378	return since, until	5✔
379
380
381	def _get_api_client(args, commcarehq_base_url):	5✔
382	return CommCareHqClient(	5✔
383	url=commcarehq_base_url,
384	project=args.project,
385	username=args.username,
386	password=args.password,
387	auth_mode=args.auth_mode,
388	version=args.api_version
389	)
390
391
392	def _get_checkpoint_manager(args):	5✔
393	if not args.users and not args.locations and not os.path.exists(	×
394	args.query
395	):
396	logger.warning(	×
397	"Checkpointing disabled for non builtin, "
398	"non file-based query"
399	)
400	elif args.since or args.until:	×
401	logger.warning(	×
402	"Checkpointing disabled when using '--since' or '--until'"
403	)
404	else:
405	checkpoint_manager = get_checkpoint_manager(args)	×
406	checkpoint_manager.create_checkpoint_table()	×
407	return checkpoint_manager	×
408
409
410	def force_lazy_result(lazy_result):	5✔
411	if lazy_result is not None:	5✔
412	if isinstance(lazy_result, RepeatableIterator):	5✔
413	list(lazy_result) if lazy_result else lazy_result	5✔
414	else:
415	for nested_result in lazy_result:	5✔
416	force_lazy_result(nested_result)	5✔
417
418
419	def evaluate_query(env, query):	5✔
420	with env:	5✔
421	try:	5✔
422	lazy_result = query.eval(env)	5✔
423	force_lazy_result(lazy_result)	5✔
424	return 0	5✔
425	except requests.exceptions.RequestException as e:	5✔
426	if e.response and e.response.status_code == 401:	×
427	print(	×
428	"\nAuthentication failed. Please check your credentials.",
429	file=sys.stderr
430	)
431	return EXIT_STATUS_ERROR	×
432	else:
433	raise	×
434	except ResourceRepeatException as e:	5✔
435	print('Stopping because the export is stuck')	×
436	print(e.message)	×
437	print('Try increasing --batch-size to overcome the error')	×
438	return EXIT_STATUS_ERROR	×
439	except (	5✔
440	sqlalchemy.exc.DataError,
441	sqlalchemy.exc.InternalError,
442	sqlalchemy.exc.ProgrammingError
443	) as e:
444	print('Stopping because of database error:\n', e)	5✔
445	return EXIT_STATUS_ERROR	5✔
446	except KeyboardInterrupt:	×
447	print('\nExport aborted', file=sys.stderr)	×
448	return EXIT_STATUS_ERROR	×
449
450
451	def main_with_args(args):	5✔
452	logger.info("CommCare Export Version {}".format(__version__))	5✔
453	writer = _get_writer(args.output_format, args.output, args.strict_types)	5✔
454
455	if args.query is None and args.users is False and args.locations is False:	5✔
456	print(	×
457	'At least one the following arguments is required: '
458	'--query, --users, --locations',
459	file=sys.stderr
460	)
461	return EXIT_STATUS_ERROR	×
462
463	if not args.username:	5✔
464	logger.warn("Username not provided")	×
465	args.username = input('Please provide a username: ')	×
466
467	if not args.password:	5✔
468	logger.warn("Password not provided")	×
469	# Windows getpass does not accept unicode
470	args.password = getpass.getpass()	×
471
472	column_enforcer = None	5✔
473	if args.with_organization:	5✔
474	column_enforcer = builtin_queries.ColumnEnforcer()	×
475
476	commcarehq_base_url = commcare_hq_aliases.get(	5✔
477	args.commcare_hq, args.commcare_hq
478	)
479	api_client = _get_api_client(args, commcarehq_base_url)	5✔
480	lp = LocationInfoProvider(api_client, page_size=args.batch_size)	5✔
481	try:	5✔
482	query = get_queries(args, writer, lp, column_enforcer)	5✔
483	except DataExportException as e:	×
484	print(e.message, file=sys.stderr)	×
485	return EXIT_STATUS_ERROR	×
486
487	if args.dump_query:	5✔
488	print(json.dumps(query.to_jvalue(), indent=4))	×
489	return	×
490
491	checkpoint_manager = None	5✔
492	if writer.support_checkpoints:	5✔
493	checkpoint_manager = _get_checkpoint_manager(args)	5✔
494
495	since, until = get_date_params(args)	5✔
496	if args.start_over:	5✔
497	if checkpoint_manager:	5✔
498	logger.warning(	5✔
499	'Ignoring all checkpoints and re-fetching all data from '
500	'CommCare.'
501	)
502	elif since:	5✔
503	logger.debug('Starting from %s', args.since)	5✔
504
505	cm = CheckpointManagerProvider(checkpoint_manager, since, args.start_over)	5!
506	static_env = {	5✔
507	'commcarehq_base_url': commcarehq_base_url,
508	'get_checkpoint_manager': cm.get_checkpoint_manager,
509	'get_location_info': lp.get_location_info,
510	'get_location_ancestor': lp.get_location_ancestor
511	}
512	env = (	5✔
513	BuiltInEnv(static_env)
514	\| CommCareHqEnv(api_client, until=until, page_size=args.batch_size)
515	\| JsonPathEnv({})
516	\| EmitterEnv(writer)
517	)
518
519	exit_status = evaluate_query(env, query)	5✔
520
521	if args.output_format == 'json':	5✔
522	print(	5✔
523	json.dumps(
524	list(writer.tables.values()),
525	indent=4,
526	default=default_to_json
527	)
528	)
529
530	return exit_status	5✔
531
532
533	def entry_point():	5✔
534	main(sys.argv[1:])	×
535
536
537	if __name__ == '__main__':	5✔
538	entry_point()	×

dimagi / commcare-export / 8559166587

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