feat: Use EDK v0.4.0 (#52)

edgarrmondragon · web-flow · commit 9a04b7b7289e · 2023-11-27T12:55:22.000-06:00
diff --git a/.flake8 b/.flake8
@@ -1,7 +1,5 @@
 [flake8]
-ignore = W503, C901, ANN101
-max-line-length = 88
+ignore = DAR
+max-line-length = 120
 exclude = files_airflow_ext,cookiecutter
-max-complexity = 10
 docstring-convention = google
-allow-star-arg-any = true
diff --git a/.github/dependabot.yml b/.github/dependabot.yml
@@ -0,0 +1,9 @@
+version: 2
+updates:
+  - package-ecosystem: "pip"
+    directory: "/"
+    schedule:
+      interval: daily
+      time: "12:00"
+      timezone: "UTC"
+    reviewers: [meltano/engineering]
diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml
@@ -1,5 +1,5 @@
 ci:
-  autofix_prs: false
+  autofix_prs: true
   autoupdate_schedule: weekly
   autoupdate_commit_msg: 'chore: pre-commit autoupdate'
 
@@ -9,38 +9,26 @@ repos:
   hooks:
   - id: check-json
   - id: check-toml
-  - id: check-yaml
+    exclude: |
+        (?x)^(
+            copier_template/.*/pyproject.toml
+        )$
   - id: end-of-file-fixer
-    exclude: (docs/.*|samples/.*\.json)
+    exclude: (copier_template/.*|docs/.*|samples/.*\.json)
   - id: trailing-whitespace
 
-- repo: https://github.com/psf/black
-  rev: 23.11.0
-  hooks:
-  - id: black
-
-- repo: https://github.com/pycqa/isort
-  rev: 5.12.0
+- repo: https://github.com/astral-sh/ruff-pre-commit
+  rev: v0.1.6
   hooks:
-  - id: isort
+  - id: ruff
+    args: [--fix, --exit-non-zero-on-fix, --show-fixes]
+  - id: ruff-format
 
 - repo: https://github.com/pycqa/flake8
   rev: 6.1.0
   hooks:
   - id: flake8
     additional_dependencies:
     - darglint==1.8.1
-    - flake8-annotations==2.9.0
-    - flake8-docstrings==1.6.0
     files: 'airflow_ext/.*'
     exclude: 'files_airflow_ext/orchestrate/meltano.py'
-
-- repo: https://github.com/asottile/pyupgrade
-  rev: v3.15.0
-  hooks:
-  - id: pyupgrade
-    args: [--py37-plus]
-    exclude: |
-        (?x)^(
-            files_airflow_ext/orchestrate/meltano.py
-        )$
diff --git a/airflow_ext/main.py b/airflow_ext/main.py
@@ -34,15 +34,11 @@ def initialize(ctx: typer.Context, force: bool = False) -> None:
     try:
         ext.initialize(force)
     except Exception:
-        log.exception(
-            "initialize failed with uncaught exception, please report to maintainer"
-        )
+        log.exception("initialize failed with uncaught exception, please report to maintainer")
         sys.exit(1)
 
 
-@app.command(
-    context_settings={"allow_extra_args": True, "ignore_unknown_options": True}
-)
+@app.command(context_settings={"allow_extra_args": True, "ignore_unknown_options": True})
 def invoke(ctx: typer.Context, command_args: List[str]) -> None:
     """Invoke the underlying wrapped cli.
 
@@ -55,17 +51,13 @@ def invoke(ctx: typer.Context, command_args: List[str]) -> None:
         command_args: The command args to invoke
     """
     command_name, command_args = command_args[0], command_args[1:]
-    log.debug(
-        "called", command_name=command_name, command_args=command_args, env=os.environ
-    )
+    log.debug("called", command_name=command_name, command_args=command_args, env=os.environ)
     ext.pass_through_invoker(log, command_name, *command_args)
 
 
 @app.command()
 def describe(
-    output_format: DescribeFormat = typer.Option(
-        DescribeFormat.text, "--format", help="Output format"
-    )
+    output_format: DescribeFormat = typer.Option(DescribeFormat.text, "--format", help="Output format"),
 ) -> None:
     """Describe the available commands of this extension.
 
@@ -75,22 +67,16 @@ def describe(
     try:
         typer.echo(ext.describe_formatted(output_format))
     except Exception:
-        log.exception(
-            "describe failed with uncaught exception, please report to maintainer"
-        )
+        log.exception("describe failed with uncaught exception, please report to maintainer")
         sys.exit(1)
 
 
 @app.callback(invoke_without_command=True)
 def main(
     ctx: typer.Context,
     log_level: str = typer.Option("INFO", envvar="LOG_LEVEL"),
-    log_timestamps: bool = typer.Option(
-        False, envvar="LOG_TIMESTAMPS", help="Show timestamp in logs"
-    ),
-    log_levels: bool = typer.Option(
-        False, "--log-levels", envvar="LOG_LEVELS", help="Show log levels"
-    ),
+    log_timestamps: bool = typer.Option(False, envvar="LOG_TIMESTAMPS", help="Show timestamp in logs"),
+    log_levels: bool = typer.Option(False, "--log-levels", envvar="LOG_LEVELS", help="Show log levels"),
     meltano_log_json: bool = typer.Option(
         False,
         "--meltano-log-json",
diff --git a/airflow_ext/pass_through.py b/airflow_ext/pass_through.py
@@ -13,5 +13,5 @@ def pass_through_cli() -> None:
     ext = Airflow()
     ext.pass_through_invoker(
         structlog.getLogger("airflow_invoker"),
-        *sys.argv[1:] if len(sys.argv) > 1 else []
+        *sys.argv[1:] if len(sys.argv) > 1 else [],
     )
diff --git a/airflow_ext/wrapper.py b/airflow_ext/wrapper.py
@@ -25,19 +25,13 @@ def __init__(self) -> None:
         self.airflow_bin = "airflow"
         self.airflow_invoker = Invoker(self.airflow_bin)
 
-        self.airflow_home = os.environ.get("AIRFLOW_HOME") or os.environ.get(
-            f"{self.app_name}_AIRFLOW_HOME"
-        )
+        self.airflow_home = os.environ.get("AIRFLOW_HOME") or os.environ.get(f"{self.app_name}_AIRFLOW_HOME")
         if not self.airflow_home:
             log.debug("env dump", env=os.environ)
-            log.error(
-                "AIRFLOW_HOME not found in environment, unable to function without it"
-            )
+            log.error("AIRFLOW_HOME not found in environment, unable to function without it")
             sys.exit(1)
 
-        self.airflow_cfg_path = Path(
-            os.environ.get("AIRFLOW_CONFIG", f"{self.airflow_home}/airflow.cfg")
-        )
+        self.airflow_cfg_path = Path(os.environ.get("AIRFLOW_CONFIG", f"{self.airflow_home}/airflow.cfg"))
         self.airflow_core_dags_path = Path(
             os.path.expandvars(
                 os.environ.get(
@@ -75,19 +69,15 @@ def initialize(self, force: bool = False) -> None:
                 "meltano dag generator not found, will be auto-generated",
                 dag_generator_path=dag_generator_path,
             )
-            dag_generator_path.write_bytes(
-                pkgutil.get_data("files_airflow_ext", "orchestrate/meltano.py")
-            )
+            dag_generator_path.write_bytes(pkgutil.get_data("files_airflow_ext", "orchestrate/meltano.py"))
 
         readme_path = self.airflow_core_dags_path / "README.md"
         if not readme_path.exists():
             log.debug(
                 "meltano dag generator README not found, will be auto-generated",
                 readme_path=readme_path,
             )
-            readme_path.write_bytes(
-                pkgutil.get_data("files_airflow_ext", "orchestrate/README.md")
-            )
+            readme_path.write_bytes(pkgutil.get_data("files_airflow_ext", "orchestrate/README.md"))
 
     def invoke(self, command_name: str | None, *command_args: Any) -> None:
         """Invoke the airflow command.
@@ -101,9 +91,7 @@ def invoke(self, command_name: str | None, *command_args: Any) -> None:
         try:
             self.airflow_invoker.run_and_log(command_name, *command_args)
         except subprocess.CalledProcessError as err:
-            log_subprocess_error(
-                f"airflow {command_name}", err, "airflow invocation failed"
-            )
+            log_subprocess_error(f"airflow {command_name}", err, "airflow invocation failed")
             sys.exit(err.returncode)
 
     def describe(self) -> models.Describe:
@@ -115,12 +103,8 @@ def describe(self) -> models.Describe:
         # TODO: could we auto-generate all or portions of this from typer instead?
         return models.Describe(
             commands=[
-                models.ExtensionCommand(
-                    name="airflow_extension", description="airflow extension commands"
-                ),
-                models.InvokerCommand(
-                    name="airflow_invoker", description="airflow pass through invoker"
-                ),
+                models.ExtensionCommand(name="airflow_extension", description="airflow extension commands"),
+                models.InvokerCommand(name="airflow_invoker", description="airflow pass through invoker"),
             ]
         )
 
@@ -131,9 +115,7 @@ def _create_config(self) -> None:
         try:
             self.airflow_invoker.run("--help", stdout=subprocess.DEVNULL)
         except subprocess.CalledProcessError as err:
-            log_subprocess_error(
-                "airflow --help", err, "initial airflow invocation failed"
-            )
+            log_subprocess_error("airflow --help", err, "initial airflow invocation failed")
             sys.exit(err.returncode)
 
     def _initdb(self) -> None:
diff --git a/files_airflow_ext/orchestrate/meltano.py b/files_airflow_ext/orchestrate/meltano.py
@@ -1,7 +1,11 @@
+"""Meltano DAG generator."""
+
 # If you want to define a custom DAG, create
 # a new file under orchestrate/dags/ and Airflow
 # will pick it up automatically.
 
+from __future__ import annotations
+
 import json
 import logging
 import os
@@ -37,13 +41,14 @@
 
 if not Path(PROJECT_ROOT).joinpath(MELTANO_BIN).exists():
     logger.warning(
-        f"A symlink to the 'meltano' executable could not be found at '{MELTANO_BIN}'. Falling back on expecting it "
-        f"to be in the PATH instead. "
+        "A symlink to the 'meltano' executable could not be found at '%s'. "
+        "Falling back on expecting it to be in the PATH instead. ",
+        MELTANO_BIN,
     )
     MELTANO_BIN = "meltano"
 
 
-def _meltano_elt_generator(schedules):
+def _meltano_elt_generator(schedules: list) -> None:
     """Generate singular dag's for each legacy Meltano elt task.
 
     Args:
@@ -53,7 +58,8 @@ def _meltano_elt_generator(schedules):
         logger.info(f"Considering schedule '{schedule['name']}': {schedule}")
         if not schedule["cron_interval"]:
             logger.info(
-                f"No DAG created for schedule '{schedule['name']}' because its interval is set to `@once`.",
+                "No DAG created for schedule '%s' because its interval is set to " "`@once`.",
+                schedule["name"],
             )
             continue
 
@@ -89,7 +95,7 @@ def _meltano_elt_generator(schedules):
             max_active_runs=1,
         )
 
-        elt = BashOperator(
+        elt = BashOperator(  # noqa: F841
             task_id="extract_load",
             bash_command=f"cd {PROJECT_ROOT}; {MELTANO_BIN} schedule run {schedule['name']}",
             dag=dag,
@@ -100,7 +106,7 @@ def _meltano_elt_generator(schedules):
         logger.info(f"DAG created for schedule '{schedule['name']}'")
 
 
-def _meltano_job_generator(schedules):
+def _meltano_job_generator(schedules: list) -> None:
     """Generate dag's for each task within a Meltano scheduled job.
 
     Args:
@@ -109,12 +115,14 @@ def _meltano_job_generator(schedules):
     for schedule in schedules:
         if not schedule.get("job"):
             logger.info(
-                f"No DAG's created for schedule '{schedule['name']}'. It was passed to job generator but has no job."
+                "No DAG's created for schedule '%s'. It was passed to job generator but has no job.",
+                schedule["name"],
             )
             continue
         if not schedule["cron_interval"]:
             logger.info(
-                f"No DAG created for schedule '{schedule['name']}' because its interval is set to `@once`."
+                "No DAG created for schedule '%s' because its interval is set to `@once`.",
+                schedule["name"],
             )
             continue
 
@@ -137,7 +145,10 @@ def _meltano_job_generator(schedules):
             previous_task = None
             for idx, task in enumerate(schedule["job"]["tasks"]):
                 logger.info(
-                    f"Considering task '{task}' of schedule '{schedule['name']}': {schedule}"
+                    "Considering task '%s' of schedule '%s': %s",
+                    task,
+                    schedule["name"],
+                    schedule,
                 )
 
                 task_id = f"{base_id}_task{idx}"
@@ -155,21 +166,19 @@ def _meltano_job_generator(schedules):
                 if previous_task:
                     task.set_upstream(previous_task)
                 previous_task = task
-                logger.info(
-                    f"Spun off task '{task}' of schedule '{schedule['name']}': {schedule}"
-                )
+                logger.info("Spun off task '%s' of schedule '%s': %s", task, schedule["name"], schedule)
 
         globals()[base_id] = dag
         logger.info(f"DAG created for schedule '{schedule['name']}', task='{run_args}'")
 
 
-def create_dags():
+def create_dags() -> None:
     """Create DAGs for Meltano schedules."""
     list_result = subprocess.run(
         [MELTANO_BIN, "schedule", "list", "--format=json"],
         cwd=PROJECT_ROOT,
         stdout=subprocess.PIPE,
-        universal_newlines=True,
+        text=True,
         check=True,
     )
     schedule_export = json.loads(list_result.stdout)
diff --git a/poetry.lock b/poetry.lock
diff --git a/pyproject.toml b/pyproject.toml

Original file line number	Diff line number	Diff line change
`@@ -13,5 +13,5 @@ def pass_through_cli() -> None:`
`13`	`13`	`ext = Airflow()`
`14`	`14`	`ext.pass_through_invoker(`
`15`	`15`	`structlog.getLogger("airflow_invoker"),`
`16`		`- *sys.argv[1:] if len(sys.argv) > 1 else []`
	`16`	`+ *sys.argv[1:] if len(sys.argv) > 1 else [],`
`17`	`17`	`)`