We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Use cleaner solution for non-truncating division in python 2
Prefer `from __future__ import division` to `1.*x/y`

--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `great_expectations/cli/v012/checkpoint.py`
Content:
```
1 import os
2 import sys
3 from typing import Dict
4 
5 import click
6 from ruamel.yaml import YAML
7 
8 from great_expectations import DataContext
9 from great_expectations.checkpoint import Checkpoint
10 from great_expectations.cli.v012 import toolkit
11 from great_expectations.cli.v012.mark import Mark as mark
12 from great_expectations.cli.v012.util import cli_message, cli_message_list
13 from great_expectations.core.expectation_suite import ExpectationSuite
14 from great_expectations.core.usage_statistics.util import send_usage_message
15 from great_expectations.data_context.types.base import DataContextConfigDefaults
16 from great_expectations.data_context.util import file_relative_path
17 from great_expectations.exceptions import InvalidTopLevelConfigKeyError
18 from great_expectations.util import lint_code
19 from great_expectations.validation_operators.types.validation_operator_result import (
20     ValidationOperatorResult,
21 )
22 
23 try:
24     from sqlalchemy.exc import SQLAlchemyError
25 except ImportError:
26     SQLAlchemyError = RuntimeError
27 
28 
29 try:
30     from sqlalchemy.exc import SQLAlchemyError
31 except ImportError:
32     SQLAlchemyError = RuntimeError
33 
34 yaml = YAML()
35 yaml.indent(mapping=2, sequence=4, offset=2)
36 
37 
38 """
39 --ge-feature-maturity-info--
40 
41     id: checkpoint_command_line
42     title: LegacyCheckpoint - Command Line
43     icon:
44     short_description: Run a configured legacy checkpoint from a command line.
45     description: Run a configured legacy checkpoint from a command line in a Terminal shell.
46     how_to_guide_url: https://docs.greatexpectations.io/en/latest/guides/how_to_guides/validation/how_to_run_a_checkpoint_in_terminal.html
47     maturity: Experimental
48     maturity_details:
49         api_stability: Unstable (expect changes to batch request; no checkpoint store)
50         implementation_completeness: Complete
51         unit_test_coverage: Complete
52         integration_infrastructure_test_coverage: N/A
53         documentation_completeness: Complete
54         bug_risk: Low
55 
56 --ge-feature-maturity-info--
57 """
58 
59 
60 @click.group(short_help="Checkpoint operations")
61 def checkpoint():
62     """
63     Checkpoint operations
64 
65     A checkpoint is a bundle of one or more batches of data with one or more
66     Expectation Suites.
67 
68     A checkpoint can be as simple as one batch of data paired with one
69     Expectation Suite.
70 
71     A checkpoint can be as complex as many batches of data across different
72     datasources paired with one or more Expectation Suites each.
73     """
74     pass
75 
76 
77 @checkpoint.command(name="new")
78 @click.argument("checkpoint")
79 @click.argument("suite")
80 @click.option("--datasource", default=None)
81 @click.option(
82     "--directory",
83     "-d",
84     default=None,
85     help="The project's great_expectations directory.",
86 )
87 @mark.cli_as_experimental
88 def checkpoint_new(checkpoint, suite, directory, datasource):
89     """Create a new checkpoint for easy deployments. (Experimental)"""
90     suite_name = suite
91     usage_event = "cli.checkpoint.new"
92     context = toolkit.load_data_context_with_error_handling(directory)
93 
94     _verify_checkpoint_does_not_exist(context, checkpoint, usage_event)
95     suite: ExpectationSuite = toolkit.load_expectation_suite(
96         context, suite_name, usage_event
97     )
98     datasource = toolkit.select_datasource(context, datasource_name=datasource)
99     if datasource is None:
100         send_usage_message(
101             data_context=context,
102             event=usage_event,
103             api_version="v2",
104             success=False,
105         )
106         sys.exit(1)
107     _, _, _, batch_kwargs = toolkit.get_batch_kwargs(context, datasource.name)
108 
109     _ = context.add_checkpoint(
110         name=checkpoint,
111         **{
112             "class_name": "LegacyCheckpoint",
113             "batches": [
114                 {
115                     "batch_kwargs": dict(batch_kwargs),
116                     "expectation_suite_names": [suite.expectation_suite_name],
117                 }
118             ],
119         },
120     )
121 
122     cli_message(
123         f"""<green>A Checkpoint named `{checkpoint}` was added to your project!</green>
124   - To run this Checkpoint, run `great_expectations checkpoint run {checkpoint}`"""
125     )
126     send_usage_message(
127         data_context=context,
128         event=usage_event,
129         api_version="v2",
130         success=True,
131     )
132 
133 
134 def _verify_checkpoint_does_not_exist(
135     context: DataContext, checkpoint: str, usage_event: str
136 ) -> None:
137     try:
138         if checkpoint in context.list_checkpoints():
139             toolkit.exit_with_failure_message_and_stats(
140                 context,
141                 usage_event,
142                 f"A checkpoint named `{checkpoint}` already exists. Please choose a new name.",
143             )
144     except InvalidTopLevelConfigKeyError as e:
145         toolkit.exit_with_failure_message_and_stats(
146             context, usage_event, f"<red>{e}</red>"
147         )
148 
149 
150 def _write_checkpoint_to_disk(
151     context: DataContext, checkpoint: Dict, checkpoint_name: str
152 ) -> str:
153     # TODO this should be the responsibility of the DataContext
154     checkpoint_dir = os.path.join(
155         context.root_directory,
156         DataContextConfigDefaults.CHECKPOINTS_BASE_DIRECTORY.value,
157     )
158     checkpoint_file = os.path.join(checkpoint_dir, f"{checkpoint_name}.yml")
159     os.makedirs(checkpoint_dir, exist_ok=True)
160     with open(checkpoint_file, "w") as f:
161         yaml.dump(checkpoint, f)
162     return checkpoint_file
163 
164 
165 def _load_checkpoint_yml_template() -> dict:
166     # TODO this should be the responsibility of the DataContext
167     template_file = file_relative_path(
168         __file__, os.path.join("..", "data_context", "checkpoint_template.yml")
169     )
170     with open(template_file) as f:
171         template = yaml.load(f)
172     return template
173 
174 
175 @checkpoint.command(name="list")
176 @click.option(
177     "--directory",
178     "-d",
179     default=None,
180     help="The project's great_expectations directory.",
181 )
182 @mark.cli_as_experimental
183 def checkpoint_list(directory):
184     """List configured checkpoints. (Experimental)"""
185     context = toolkit.load_data_context_with_error_handling(directory)
186     checkpoints = context.list_checkpoints()
187     if not checkpoints:
188         cli_message(
189             "No checkpoints found.\n"
190             "  - Use the command `great_expectations checkpoint new` to create one."
191         )
192         send_usage_message(
193             data_context=context,
194             event="cli.checkpoint.list",
195             api_version="v2",
196             success=True,
197         )
198         sys.exit(0)
199 
200     number_found = len(checkpoints)
201     plural = "s" if number_found > 1 else ""
202     message = f"Found {number_found} checkpoint{plural}."
203     pretty_list = [f" - <cyan>{cp}</cyan>" for cp in checkpoints]
204     cli_message_list(pretty_list, list_intro_string=message)
205     send_usage_message(
206         data_context=context,
207         event="cli.checkpoint.list",
208         api_version="v2",
209         success=True,
210     )
211 
212 
213 @checkpoint.command(name="run")
214 @click.argument("checkpoint")
215 @click.option(
216     "--directory",
217     "-d",
218     default=None,
219     help="The project's great_expectations directory.",
220 )
221 @mark.cli_as_experimental
222 def checkpoint_run(checkpoint, directory):
223     """Run a checkpoint. (Experimental)"""
224     usage_event = "cli.checkpoint.run"
225     context = toolkit.load_data_context_with_error_handling(
226         directory=directory, from_cli_upgrade_command=False
227     )
228 
229     checkpoint: Checkpoint = toolkit.load_checkpoint(
230         context,
231         checkpoint,
232         usage_event,
233     )
234 
235     try:
236         results = checkpoint.run()
237     except Exception as e:
238         toolkit.exit_with_failure_message_and_stats(
239             context, usage_event, f"<red>{e}</red>"
240         )
241 
242     if not results["success"]:
243         cli_message("Validation failed!")
244         send_usage_message(
245             data_context=context,
246             event=usage_event,
247             api_version="v2",
248             success=True,
249         )
250         print_validation_operator_results_details(results)
251         sys.exit(1)
252 
253     cli_message("Validation succeeded!")
254     send_usage_message(
255         data_context=context,
256         event=usage_event,
257         api_version="v2",
258         success=True,
259     )
260     print_validation_operator_results_details(results)
261     sys.exit(0)
262 
263 
264 def print_validation_operator_results_details(
265     results: ValidationOperatorResult,
266 ) -> None:
267     max_suite_display_width = 40
268     toolkit.cli_message(
269         f"""
270 {'Suite Name'.ljust(max_suite_display_width)}     Status     Expectations met"""
271     )
272     for id, result in results.run_results.items():
273         vr = result["validation_result"]
274         stats = vr.statistics
275         passed = stats["successful_expectations"]
276         evaluated = stats["evaluated_expectations"]
277         percentage_slug = (
278             f"{round(passed / evaluated * 100, 2) if evaluated > 0 else 100} %"
279         )
280         stats_slug = f"{passed} of {evaluated} ({percentage_slug})"
281         if vr.success:
282             status_slug = "<green>✔ Passed</green>"
283         else:
284             status_slug = "<red>✖ Failed</red>"
285         suite_name = str(vr.meta["expectation_suite_name"])
286         if len(suite_name) > max_suite_display_width:
287             suite_name = suite_name[0:max_suite_display_width]
288             suite_name = suite_name[:-1] + "…"
289         status_line = f"- {suite_name.ljust(max_suite_display_width)}   {status_slug}   {stats_slug}"
290         toolkit.cli_message(status_line)
291 
292 
293 @checkpoint.command(name="script")
294 @click.argument("checkpoint")
295 @click.option(
296     "--directory",
297     "-d",
298     default=None,
299     help="The project's great_expectations directory.",
300 )
301 @mark.cli_as_experimental
302 def checkpoint_script(checkpoint, directory):
303     """
304     Create a python script to run a checkpoint. (Experimental)
305 
306     Checkpoints can be run directly without this script using the
307     `great_expectations checkpoint run` command.
308 
309     This script is provided for those who wish to run checkpoints via python.
310     """
311     context = toolkit.load_data_context_with_error_handling(directory)
312     usage_event = "cli.checkpoint.script"
313 
314     # Attempt to load the checkpoint and deal with errors
315     _ = toolkit.load_checkpoint(context, checkpoint, usage_event)
316 
317     script_name = f"run_{checkpoint}.py"
318     script_path = os.path.join(
319         context.root_directory, context.GE_UNCOMMITTED_DIR, script_name
320     )
321 
322     if os.path.isfile(script_path):
323         toolkit.exit_with_failure_message_and_stats(
324             context,
325             usage_event,
326             f"""<red>Warning! A script named {script_name} already exists and this command will not overwrite it.</red>
327   - Existing file path: {script_path}""",
328         )
329 
330     _write_checkpoint_script_to_disk(context.root_directory, checkpoint, script_path)
331     cli_message(
332         f"""<green>A python script was created that runs the checkpoint named: `{checkpoint}`</green>
333   - The script is located in `great_expectations/uncommitted/run_{checkpoint}.py`
334   - The script can be run with `python great_expectations/uncommitted/run_{checkpoint}.py`"""
335     )
336     send_usage_message(
337         data_context=context,
338         event=usage_event,
339         api_version="v2",
340         success=True,
341     )
342 
343 
344 def _load_script_template() -> str:
345     with open(file_relative_path(__file__, "checkpoint_script_template.py")) as f:
346         template = f.read()
347     return template
348 
349 
350 def _write_checkpoint_script_to_disk(
351     context_directory: str, checkpoint_name: str, script_path: str
352 ) -> None:
353     script_full_path = os.path.abspath(os.path.join(script_path))
354     template = _load_script_template().format(checkpoint_name, context_directory)
355     linted_code = lint_code(template)
356     with open(script_full_path, "w") as f:
357         f.write(linted_code)
358
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/great_expectations/cli/v012/checkpoint.py b/great_expectations/cli/v012/checkpoint.py
--- a/great_expectations/cli/v012/checkpoint.py
+++ b/great_expectations/cli/v012/checkpoint.py
@@ -121,7 +121,7 @@
 
     cli_message(
         f"""<green>A Checkpoint named `{checkpoint}` was added to your project!</green>
-  - To run this Checkpoint, run `great_expectations checkpoint run {checkpoint}`"""
+  - To run this Checkpoint, run `great_expectations --v2-api checkpoint run {checkpoint}`"""
     )
     send_usage_message(
         data_context=context,

{"golden_diff": "diff --git a/great_expectations/cli/v012/checkpoint.py b/great_expectations/cli/v012/checkpoint.py\n--- a/great_expectations/cli/v012/checkpoint.py\n+++ b/great_expectations/cli/v012/checkpoint.py\n@@ -121,7 +121,7 @@\n \n     cli_message(\n         f\"\"\"<green>A Checkpoint named `{checkpoint}` was added to your project!</green>\n-  - To run this Checkpoint, run `great_expectations checkpoint run {checkpoint}`\"\"\"\n+  - To run this Checkpoint, run `great_expectations --v2-api checkpoint run {checkpoint}`\"\"\"\n     )\n     send_usage_message(\n         data_context=context,\n", "issue": "Use cleaner solution for non-truncating division in python 2\nPrefer `from __future__ import division` to `1.*x/y`\n", "before_files": [{"content": "import os\nimport sys\nfrom typing import Dict\n\nimport click\nfrom ruamel.yaml import YAML\n\nfrom great_expectations import DataContext\nfrom great_expectations.checkpoint import Checkpoint\nfrom great_expectations.cli.v012 import toolkit\nfrom great_expectations.cli.v012.mark import Mark as mark\nfrom great_expectations.cli.v012.util import cli_message, cli_message_list\nfrom great_expectations.core.expectation_suite import ExpectationSuite\nfrom great_expectations.core.usage_statistics.util import send_usage_message\nfrom great_expectations.data_context.types.base import DataContextConfigDefaults\nfrom great_expectations.data_context.util import file_relative_path\nfrom great_expectations.exceptions import InvalidTopLevelConfigKeyError\nfrom great_expectations.util import lint_code\nfrom great_expectations.validation_operators.types.validation_operator_result import (\n    ValidationOperatorResult,\n)\n\ntry:\n    from sqlalchemy.exc import SQLAlchemyError\nexcept ImportError:\n    SQLAlchemyError = RuntimeError\n\n\ntry:\n    from sqlalchemy.exc import SQLAlchemyError\nexcept ImportError:\n    SQLAlchemyError = RuntimeError\n\nyaml = YAML()\nyaml.indent(mapping=2, sequence=4, offset=2)\n\n\n\"\"\"\n--ge-feature-maturity-info--\n\n    id: checkpoint_command_line\n    title: LegacyCheckpoint - Command Line\n    icon:\n    short_description: Run a configured legacy checkpoint from a command line.\n    description: Run a configured legacy checkpoint from a command line in a Terminal shell.\n    how_to_guide_url: https://docs.greatexpectations.io/en/latest/guides/how_to_guides/validation/how_to_run_a_checkpoint_in_terminal.html\n    maturity: Experimental\n    maturity_details:\n        api_stability: Unstable (expect changes to batch request; no checkpoint store)\n        implementation_completeness: Complete\n        unit_test_coverage: Complete\n        integration_infrastructure_test_coverage: N/A\n        documentation_completeness: Complete\n        bug_risk: Low\n\n--ge-feature-maturity-info--\n\"\"\"\n\n\n@click.group(short_help=\"Checkpoint operations\")\ndef checkpoint():\n    \"\"\"\n    Checkpoint operations\n\n    A checkpoint is a bundle of one or more batches of data with one or more\n    Expectation Suites.\n\n    A checkpoint can be as simple as one batch of data paired with one\n    Expectation Suite.\n\n    A checkpoint can be as complex as many batches of data across different\n    datasources paired with one or more Expectation Suites each.\n    \"\"\"\n    pass\n\n\n@checkpoint.command(name=\"new\")\n@click.argument(\"checkpoint\")\n@click.argument(\"suite\")\n@click.option(\"--datasource\", default=None)\n@click.option(\n    \"--directory\",\n    \"-d\",\n    default=None,\n    help=\"The project's great_expectations directory.\",\n)\n@mark.cli_as_experimental\ndef checkpoint_new(checkpoint, suite, directory, datasource):\n    \"\"\"Create a new checkpoint for easy deployments. (Experimental)\"\"\"\n    suite_name = suite\n    usage_event = \"cli.checkpoint.new\"\n    context = toolkit.load_data_context_with_error_handling(directory)\n\n    _verify_checkpoint_does_not_exist(context, checkpoint, usage_event)\n    suite: ExpectationSuite = toolkit.load_expectation_suite(\n        context, suite_name, usage_event\n    )\n    datasource = toolkit.select_datasource(context, datasource_name=datasource)\n    if datasource is None:\n        send_usage_message(\n            data_context=context,\n            event=usage_event,\n            api_version=\"v2\",\n            success=False,\n        )\n        sys.exit(1)\n    _, _, _, batch_kwargs = toolkit.get_batch_kwargs(context, datasource.name)\n\n    _ = context.add_checkpoint(\n        name=checkpoint,\n        **{\n            \"class_name\": \"LegacyCheckpoint\",\n            \"batches\": [\n                {\n                    \"batch_kwargs\": dict(batch_kwargs),\n                    \"expectation_suite_names\": [suite.expectation_suite_name],\n                }\n            ],\n        },\n    )\n\n    cli_message(\n        f\"\"\"<green>A Checkpoint named `{checkpoint}` was added to your project!</green>\n  - To run this Checkpoint, run `great_expectations checkpoint run {checkpoint}`\"\"\"\n    )\n    send_usage_message(\n        data_context=context,\n        event=usage_event,\n        api_version=\"v2\",\n        success=True,\n    )\n\n\ndef _verify_checkpoint_does_not_exist(\n    context: DataContext, checkpoint: str, usage_event: str\n) -> None:\n    try:\n        if checkpoint in context.list_checkpoints():\n            toolkit.exit_with_failure_message_and_stats(\n                context,\n                usage_event,\n                f\"A checkpoint named `{checkpoint}` already exists. Please choose a new name.\",\n            )\n    except InvalidTopLevelConfigKeyError as e:\n        toolkit.exit_with_failure_message_and_stats(\n            context, usage_event, f\"<red>{e}</red>\"\n        )\n\n\ndef _write_checkpoint_to_disk(\n    context: DataContext, checkpoint: Dict, checkpoint_name: str\n) -> str:\n    # TODO this should be the responsibility of the DataContext\n    checkpoint_dir = os.path.join(\n        context.root_directory,\n        DataContextConfigDefaults.CHECKPOINTS_BASE_DIRECTORY.value,\n    )\n    checkpoint_file = os.path.join(checkpoint_dir, f\"{checkpoint_name}.yml\")\n    os.makedirs(checkpoint_dir, exist_ok=True)\n    with open(checkpoint_file, \"w\") as f:\n        yaml.dump(checkpoint, f)\n    return checkpoint_file\n\n\ndef _load_checkpoint_yml_template() -> dict:\n    # TODO this should be the responsibility of the DataContext\n    template_file = file_relative_path(\n        __file__, os.path.join(\"..\", \"data_context\", \"checkpoint_template.yml\")\n    )\n    with open(template_file) as f:\n        template = yaml.load(f)\n    return template\n\n\n@checkpoint.command(name=\"list\")\n@click.option(\n    \"--directory\",\n    \"-d\",\n    default=None,\n    help=\"The project's great_expectations directory.\",\n)\n@mark.cli_as_experimental\ndef checkpoint_list(directory):\n    \"\"\"List configured checkpoints. (Experimental)\"\"\"\n    context = toolkit.load_data_context_with_error_handling(directory)\n    checkpoints = context.list_checkpoints()\n    if not checkpoints:\n        cli_message(\n            \"No checkpoints found.\\n\"\n            \"  - Use the command `great_expectations checkpoint new` to create one.\"\n        )\n        send_usage_message(\n            data_context=context,\n            event=\"cli.checkpoint.list\",\n            api_version=\"v2\",\n            success=True,\n        )\n        sys.exit(0)\n\n    number_found = len(checkpoints)\n    plural = \"s\" if number_found > 1 else \"\"\n    message = f\"Found {number_found} checkpoint{plural}.\"\n    pretty_list = [f\" - <cyan>{cp}</cyan>\" for cp in checkpoints]\n    cli_message_list(pretty_list, list_intro_string=message)\n    send_usage_message(\n        data_context=context,\n        event=\"cli.checkpoint.list\",\n        api_version=\"v2\",\n        success=True,\n    )\n\n\n@checkpoint.command(name=\"run\")\n@click.argument(\"checkpoint\")\n@click.option(\n    \"--directory\",\n    \"-d\",\n    default=None,\n    help=\"The project's great_expectations directory.\",\n)\n@mark.cli_as_experimental\ndef checkpoint_run(checkpoint, directory):\n    \"\"\"Run a checkpoint. (Experimental)\"\"\"\n    usage_event = \"cli.checkpoint.run\"\n    context = toolkit.load_data_context_with_error_handling(\n        directory=directory, from_cli_upgrade_command=False\n    )\n\n    checkpoint: Checkpoint = toolkit.load_checkpoint(\n        context,\n        checkpoint,\n        usage_event,\n    )\n\n    try:\n        results = checkpoint.run()\n    except Exception as e:\n        toolkit.exit_with_failure_message_and_stats(\n            context, usage_event, f\"<red>{e}</red>\"\n        )\n\n    if not results[\"success\"]:\n        cli_message(\"Validation failed!\")\n        send_usage_message(\n            data_context=context,\n            event=usage_event,\n            api_version=\"v2\",\n            success=True,\n        )\n        print_validation_operator_results_details(results)\n        sys.exit(1)\n\n    cli_message(\"Validation succeeded!\")\n    send_usage_message(\n        data_context=context,\n        event=usage_event,\n        api_version=\"v2\",\n        success=True,\n    )\n    print_validation_operator_results_details(results)\n    sys.exit(0)\n\n\ndef print_validation_operator_results_details(\n    results: ValidationOperatorResult,\n) -> None:\n    max_suite_display_width = 40\n    toolkit.cli_message(\n        f\"\"\"\n{'Suite Name'.ljust(max_suite_display_width)}     Status     Expectations met\"\"\"\n    )\n    for id, result in results.run_results.items():\n        vr = result[\"validation_result\"]\n        stats = vr.statistics\n        passed = stats[\"successful_expectations\"]\n        evaluated = stats[\"evaluated_expectations\"]\n        percentage_slug = (\n            f\"{round(passed / evaluated * 100, 2) if evaluated > 0 else 100} %\"\n        )\n        stats_slug = f\"{passed} of {evaluated} ({percentage_slug})\"\n        if vr.success:\n            status_slug = \"<green>\u2714 Passed</green>\"\n        else:\n            status_slug = \"<red>\u2716 Failed</red>\"\n        suite_name = str(vr.meta[\"expectation_suite_name\"])\n        if len(suite_name) > max_suite_display_width:\n            suite_name = suite_name[0:max_suite_display_width]\n            suite_name = suite_name[:-1] + \"\u2026\"\n        status_line = f\"- {suite_name.ljust(max_suite_display_width)}   {status_slug}   {stats_slug}\"\n        toolkit.cli_message(status_line)\n\n\n@checkpoint.command(name=\"script\")\n@click.argument(\"checkpoint\")\n@click.option(\n    \"--directory\",\n    \"-d\",\n    default=None,\n    help=\"The project's great_expectations directory.\",\n)\n@mark.cli_as_experimental\ndef checkpoint_script(checkpoint, directory):\n    \"\"\"\n    Create a python script to run a checkpoint. (Experimental)\n\n    Checkpoints can be run directly without this script using the\n    `great_expectations checkpoint run` command.\n\n    This script is provided for those who wish to run checkpoints via python.\n    \"\"\"\n    context = toolkit.load_data_context_with_error_handling(directory)\n    usage_event = \"cli.checkpoint.script\"\n\n    # Attempt to load the checkpoint and deal with errors\n    _ = toolkit.load_checkpoint(context, checkpoint, usage_event)\n\n    script_name = f\"run_{checkpoint}.py\"\n    script_path = os.path.join(\n        context.root_directory, context.GE_UNCOMMITTED_DIR, script_name\n    )\n\n    if os.path.isfile(script_path):\n        toolkit.exit_with_failure_message_and_stats(\n            context,\n            usage_event,\n            f\"\"\"<red>Warning! A script named {script_name} already exists and this command will not overwrite it.</red>\n  - Existing file path: {script_path}\"\"\",\n        )\n\n    _write_checkpoint_script_to_disk(context.root_directory, checkpoint, script_path)\n    cli_message(\n        f\"\"\"<green>A python script was created that runs the checkpoint named: `{checkpoint}`</green>\n  - The script is located in `great_expectations/uncommitted/run_{checkpoint}.py`\n  - The script can be run with `python great_expectations/uncommitted/run_{checkpoint}.py`\"\"\"\n    )\n    send_usage_message(\n        data_context=context,\n        event=usage_event,\n        api_version=\"v2\",\n        success=True,\n    )\n\n\ndef _load_script_template() -> str:\n    with open(file_relative_path(__file__, \"checkpoint_script_template.py\")) as f:\n        template = f.read()\n    return template\n\n\ndef _write_checkpoint_script_to_disk(\n    context_directory: str, checkpoint_name: str, script_path: str\n) -> None:\n    script_full_path = os.path.abspath(os.path.join(script_path))\n    template = _load_script_template().format(checkpoint_name, context_directory)\n    linted_code = lint_code(template)\n    with open(script_full_path, \"w\") as f:\n        f.write(linted_code)\n", "path": "great_expectations/cli/v012/checkpoint.py"}], "after_files": [{"content": "import os\nimport sys\nfrom typing import Dict\n\nimport click\nfrom ruamel.yaml import YAML\n\nfrom great_expectations import DataContext\nfrom great_expectations.checkpoint import Checkpoint\nfrom great_expectations.cli.v012 import toolkit\nfrom great_expectations.cli.v012.mark import Mark as mark\nfrom great_expectations.cli.v012.util import cli_message, cli_message_list\nfrom great_expectations.core.expectation_suite import ExpectationSuite\nfrom great_expectations.core.usage_statistics.util import send_usage_message\nfrom great_expectations.data_context.types.base import DataContextConfigDefaults\nfrom great_expectations.data_context.util import file_relative_path\nfrom great_expectations.exceptions import InvalidTopLevelConfigKeyError\nfrom great_expectations.util import lint_code\nfrom great_expectations.validation_operators.types.validation_operator_result import (\n    ValidationOperatorResult,\n)\n\ntry:\n    from sqlalchemy.exc import SQLAlchemyError\nexcept ImportError:\n    SQLAlchemyError = RuntimeError\n\n\ntry:\n    from sqlalchemy.exc import SQLAlchemyError\nexcept ImportError:\n    SQLAlchemyError = RuntimeError\n\nyaml = YAML()\nyaml.indent(mapping=2, sequence=4, offset=2)\n\n\n\"\"\"\n--ge-feature-maturity-info--\n\n    id: checkpoint_command_line\n    title: LegacyCheckpoint - Command Line\n    icon:\n    short_description: Run a configured legacy checkpoint from a command line.\n    description: Run a configured legacy checkpoint from a command line in a Terminal shell.\n    how_to_guide_url: https://docs.greatexpectations.io/en/latest/guides/how_to_guides/validation/how_to_run_a_checkpoint_in_terminal.html\n    maturity: Experimental\n    maturity_details:\n        api_stability: Unstable (expect changes to batch request; no checkpoint store)\n        implementation_completeness: Complete\n        unit_test_coverage: Complete\n        integration_infrastructure_test_coverage: N/A\n        documentation_completeness: Complete\n        bug_risk: Low\n\n--ge-feature-maturity-info--\n\"\"\"\n\n\n@click.group(short_help=\"Checkpoint operations\")\ndef checkpoint():\n    \"\"\"\n    Checkpoint operations\n\n    A checkpoint is a bundle of one or more batches of data with one or more\n    Expectation Suites.\n\n    A checkpoint can be as simple as one batch of data paired with one\n    Expectation Suite.\n\n    A checkpoint can be as complex as many batches of data across different\n    datasources paired with one or more Expectation Suites each.\n    \"\"\"\n    pass\n\n\n@checkpoint.command(name=\"new\")\n@click.argument(\"checkpoint\")\n@click.argument(\"suite\")\n@click.option(\"--datasource\", default=None)\n@click.option(\n    \"--directory\",\n    \"-d\",\n    default=None,\n    help=\"The project's great_expectations directory.\",\n)\n@mark.cli_as_experimental\ndef checkpoint_new(checkpoint, suite, directory, datasource):\n    \"\"\"Create a new checkpoint for easy deployments. (Experimental)\"\"\"\n    suite_name = suite\n    usage_event = \"cli.checkpoint.new\"\n    context = toolkit.load_data_context_with_error_handling(directory)\n\n    _verify_checkpoint_does_not_exist(context, checkpoint, usage_event)\n    suite: ExpectationSuite = toolkit.load_expectation_suite(\n        context, suite_name, usage_event\n    )\n    datasource = toolkit.select_datasource(context, datasource_name=datasource)\n    if datasource is None:\n        send_usage_message(\n            data_context=context,\n            event=usage_event,\n            api_version=\"v2\",\n            success=False,\n        )\n        sys.exit(1)\n    _, _, _, batch_kwargs = toolkit.get_batch_kwargs(context, datasource.name)\n\n    _ = context.add_checkpoint(\n        name=checkpoint,\n        **{\n            \"class_name\": \"LegacyCheckpoint\",\n            \"batches\": [\n                {\n                    \"batch_kwargs\": dict(batch_kwargs),\n                    \"expectation_suite_names\": [suite.expectation_suite_name],\n                }\n            ],\n        },\n    )\n\n    cli_message(\n        f\"\"\"<green>A Checkpoint named `{checkpoint}` was added to your project!</green>\n  - To run this Checkpoint, run `great_expectations --v2-api checkpoint run {checkpoint}`\"\"\"\n    )\n    send_usage_message(\n        data_context=context,\n        event=usage_event,\n        api_version=\"v2\",\n        success=True,\n    )\n\n\ndef _verify_checkpoint_does_not_exist(\n    context: DataContext, checkpoint: str, usage_event: str\n) -> None:\n    try:\n        if checkpoint in context.list_checkpoints():\n            toolkit.exit_with_failure_message_and_stats(\n                context,\n                usage_event,\n                f\"A checkpoint named `{checkpoint}` already exists. Please choose a new name.\",\n            )\n    except InvalidTopLevelConfigKeyError as e:\n        toolkit.exit_with_failure_message_and_stats(\n            context, usage_event, f\"<red>{e}</red>\"\n        )\n\n\ndef _write_checkpoint_to_disk(\n    context: DataContext, checkpoint: Dict, checkpoint_name: str\n) -> str:\n    # TODO this should be the responsibility of the DataContext\n    checkpoint_dir = os.path.join(\n        context.root_directory,\n        DataContextConfigDefaults.CHECKPOINTS_BASE_DIRECTORY.value,\n    )\n    checkpoint_file = os.path.join(checkpoint_dir, f\"{checkpoint_name}.yml\")\n    os.makedirs(checkpoint_dir, exist_ok=True)\n    with open(checkpoint_file, \"w\") as f:\n        yaml.dump(checkpoint, f)\n    return checkpoint_file\n\n\ndef _load_checkpoint_yml_template() -> dict:\n    # TODO this should be the responsibility of the DataContext\n    template_file = file_relative_path(\n        __file__, os.path.join(\"..\", \"data_context\", \"checkpoint_template.yml\")\n    )\n    with open(template_file) as f:\n        template = yaml.load(f)\n    return template\n\n\n@checkpoint.command(name=\"list\")\n@click.option(\n    \"--directory\",\n    \"-d\",\n    default=None,\n    help=\"The project's great_expectations directory.\",\n)\n@mark.cli_as_experimental\ndef checkpoint_list(directory):\n    \"\"\"List configured checkpoints. (Experimental)\"\"\"\n    context = toolkit.load_data_context_with_error_handling(directory)\n    checkpoints = context.list_checkpoints()\n    if not checkpoints:\n        cli_message(\n            \"No checkpoints found.\\n\"\n            \"  - Use the command `great_expectations checkpoint new` to create one.\"\n        )\n        send_usage_message(\n            data_context=context,\n            event=\"cli.checkpoint.list\",\n            api_version=\"v2\",\n            success=True,\n        )\n        sys.exit(0)\n\n    number_found = len(checkpoints)\n    plural = \"s\" if number_found > 1 else \"\"\n    message = f\"Found {number_found} checkpoint{plural}.\"\n    pretty_list = [f\" - <cyan>{cp}</cyan>\" for cp in checkpoints]\n    cli_message_list(pretty_list, list_intro_string=message)\n    send_usage_message(\n        data_context=context,\n        event=\"cli.checkpoint.list\",\n        api_version=\"v2\",\n        success=True,\n    )\n\n\n@checkpoint.command(name=\"run\")\n@click.argument(\"checkpoint\")\n@click.option(\n    \"--directory\",\n    \"-d\",\n    default=None,\n    help=\"The project's great_expectations directory.\",\n)\n@mark.cli_as_experimental\ndef checkpoint_run(checkpoint, directory):\n    \"\"\"Run a checkpoint. (Experimental)\"\"\"\n    usage_event = \"cli.checkpoint.run\"\n    context = toolkit.load_data_context_with_error_handling(\n        directory=directory, from_cli_upgrade_command=False\n    )\n\n    checkpoint: Checkpoint = toolkit.load_checkpoint(\n        context,\n        checkpoint,\n        usage_event,\n    )\n\n    try:\n        results = checkpoint.run()\n    except Exception as e:\n        toolkit.exit_with_failure_message_and_stats(\n            context, usage_event, f\"<red>{e}</red>\"\n        )\n\n    if not results[\"success\"]:\n        cli_message(\"Validation failed!\")\n        send_usage_message(\n            data_context=context,\n            event=usage_event,\n            api_version=\"v2\",\n            success=True,\n        )\n        print_validation_operator_results_details(results)\n        sys.exit(1)\n\n    cli_message(\"Validation succeeded!\")\n    send_usage_message(\n        data_context=context,\n        event=usage_event,\n        api_version=\"v2\",\n        success=True,\n    )\n    print_validation_operator_results_details(results)\n    sys.exit(0)\n\n\ndef print_validation_operator_results_details(\n    results: ValidationOperatorResult,\n) -> None:\n    max_suite_display_width = 40\n    toolkit.cli_message(\n        f\"\"\"\n{'Suite Name'.ljust(max_suite_display_width)}     Status     Expectations met\"\"\"\n    )\n    for id, result in results.run_results.items():\n        vr = result[\"validation_result\"]\n        stats = vr.statistics\n        passed = stats[\"successful_expectations\"]\n        evaluated = stats[\"evaluated_expectations\"]\n        percentage_slug = (\n            f\"{round(passed / evaluated * 100, 2) if evaluated > 0 else 100} %\"\n        )\n        stats_slug = f\"{passed} of {evaluated} ({percentage_slug})\"\n        if vr.success:\n            status_slug = \"<green>\u2714 Passed</green>\"\n        else:\n            status_slug = \"<red>\u2716 Failed</red>\"\n        suite_name = str(vr.meta[\"expectation_suite_name\"])\n        if len(suite_name) > max_suite_display_width:\n            suite_name = suite_name[0:max_suite_display_width]\n            suite_name = suite_name[:-1] + \"\u2026\"\n        status_line = f\"- {suite_name.ljust(max_suite_display_width)}   {status_slug}   {stats_slug}\"\n        toolkit.cli_message(status_line)\n\n\n@checkpoint.command(name=\"script\")\n@click.argument(\"checkpoint\")\n@click.option(\n    \"--directory\",\n    \"-d\",\n    default=None,\n    help=\"The project's great_expectations directory.\",\n)\n@mark.cli_as_experimental\ndef checkpoint_script(checkpoint, directory):\n    \"\"\"\n    Create a python script to run a checkpoint. (Experimental)\n\n    Checkpoints can be run directly without this script using the\n    `great_expectations checkpoint run` command.\n\n    This script is provided for those who wish to run checkpoints via python.\n    \"\"\"\n    context = toolkit.load_data_context_with_error_handling(directory)\n    usage_event = \"cli.checkpoint.script\"\n\n    # Attempt to load the checkpoint and deal with errors\n    _ = toolkit.load_checkpoint(context, checkpoint, usage_event)\n\n    script_name = f\"run_{checkpoint}.py\"\n    script_path = os.path.join(\n        context.root_directory, context.GE_UNCOMMITTED_DIR, script_name\n    )\n\n    if os.path.isfile(script_path):\n        toolkit.exit_with_failure_message_and_stats(\n            context,\n            usage_event,\n            f\"\"\"<red>Warning! A script named {script_name} already exists and this command will not overwrite it.</red>\n  - Existing file path: {script_path}\"\"\",\n        )\n\n    _write_checkpoint_script_to_disk(context.root_directory, checkpoint, script_path)\n    cli_message(\n        f\"\"\"<green>A python script was created that runs the checkpoint named: `{checkpoint}`</green>\n  - The script is located in `great_expectations/uncommitted/run_{checkpoint}.py`\n  - The script can be run with `python great_expectations/uncommitted/run_{checkpoint}.py`\"\"\"\n    )\n    send_usage_message(\n        data_context=context,\n        event=usage_event,\n        api_version=\"v2\",\n        success=True,\n    )\n\n\ndef _load_script_template() -> str:\n    with open(file_relative_path(__file__, \"checkpoint_script_template.py\")) as f:\n        template = f.read()\n    return template\n\n\ndef _write_checkpoint_script_to_disk(\n    context_directory: str, checkpoint_name: str, script_path: str\n) -> None:\n    script_full_path = os.path.abspath(os.path.join(script_path))\n    template = _load_script_template().format(checkpoint_name, context_directory)\n    linted_code = lint_code(template)\n    with open(script_full_path, \"w\") as f:\n        f.write(linted_code)\n", "path": "great_expectations/cli/v012/checkpoint.py"}]}