Pytest with f2s ops

.github/workflows/pytest.yml

@@ -8,11 +8,18 @@ on:
       - opened
       - synchronize
       - ready_for_review
+concurrency:
+  group: ${{ github.workflow }}-${{ github.ref }}-${{ github.event_name }}
+  cancel-in-progress: true
 
 env:
   PUDL_OUTPUT: /home/runner/pudl-work/output/
   PUDL_INPUT: /home/runner/pudl-work/input/
   DAGSTER_HOME: /home/runner/pudl-work/dagster_home/
+  ETL_CONFIG: src/pudl/package_data/settings/etl_fast.yml
+  ETL_COMMANDLINE_OPTIONS: --gcs-cache-path=gs://zenodo-cache.catalyst.coop
+  COVERAGE_OPTIONS: --concurrency=multiprocessing
+  XBRL_WORKERS: 2
 
 jobs:
   ci-docs:
@@ -64,7 +71,6 @@ jobs:
     defaults:
       run:
         shell: bash -l {0}
-
     steps:
       - uses: actions/checkout@v4
         with:
@@ -105,31 +111,36 @@ jobs:
           path: coverage.xml
 
   ci-integration:
-    runs-on:
-      group: large-runner-group
-      labels: ubuntu-22.04-4core
+    needs:
+      - ci-unit
+    runs-on: ubuntu-22.04-4core
     if: github.event.pull_request.draft == false
     permissions:
       contents: read
       id-token: write
-    strategy:
-      fail-fast: false
     defaults:
       run:
         shell: bash -l {0}
-
+    strategy:
+      fail-fast: false
+      matrix:
+        dagster_workers: [0, 1, 2, 4]
+        xbrl_workers: [1, 2, 3, 4, 8]
     steps:
       - uses: actions/checkout@v4
         with:
           fetch-depth: 2
 
-      - name: Install Conda environment using mamba
+      - name: Install conda-lock environment with micromamba
         uses: mamba-org/setup-micromamba@v1
         with:
           environment-file: environments/conda-lock.yml
           environment-name: pudl-dev
           cache-environment: true
 
+      - name: Install PUDL and its dependencies
+        run: pip install --no-deps --no-cache-dir .
+
       - name: Log environment details
         run: |
           conda info
@@ -169,21 +180,42 @@ jobs:
           workload_identity_provider: "projects/345950277072/locations/global/workloadIdentityPools/gh-actions-pool/providers/gh-actions-provider"
           service_account: "tox-pytest-github-action@catalyst-cooperative-pudl.iam.gserviceaccount.com"
 
-      - name: Run integration tests, trying to use GCS cache if possible
+      - name: Run ferc_to_sqlite
+        env:
+          COVERAGE_FILE: .coverage.ferc_to_sqlite
         run: |
-          pip install --no-deps --editable .
-          make pytest-integration
-
+          coverage run ${{ env.COVERAGE_OPTIONS }} \
+            src/pudl/ferc_to_sqlite/cli.py \
+              --dagster-workers ${{ matrix.dagster_workers }} \
+              --workers ${{ matrix.xbrl_workers }} \
+              --clobber ${{ env.ETL_COMMANDLINE_OPTIONS }} ${{ env.ETL_CONFIG }}
+      - name: Fail pipeline early
+        run: exit 1
+      - name: Run pudl_etl
+        env:
+          COVERAGE_FILE: .coverage.pudl_etl
+        run: |
+          alembic upgrade head
+          coverage run ${{ env.COVERAGE_OPTIONS }} \
+            src/pudl/cli/etl.py ${{ env.ETL_COMMANDLINE_OPTIONS }} ${{ env.ETL_CONFIG }}
+      - name: Run integration tests
+        env:
+          COVERAGE_FILE: .coverage.pytest
+        run: |
+          coverage run ${{ env.COVERAGE_OPTIONS }} \
+           -m pytest -n auto --live-dbs test/integration
+      - name: Generate coverage
+        run: |
+          coverage combine
+          coverage xml
       - name: Upload coverage
         uses: actions/upload-artifact@v3
         with:
           name: coverage-integration
           path: coverage.xml
 
-      - name: Log post-test Zenodo datastore contents
-        run: find ${{ env.PUDL_INPUT }}
-
   ci-coverage:
+    name: Upload coverage to CodeCov
     runs-on: ubuntu-latest
     needs:
       - ci-docs
@@ -197,37 +229,40 @@ jobs:
         with:
           path: coverage
       - name: List downloaded files
-        run: |
-          ls -R
+        run: find -type f
       - name: Upload test coverage report to CodeCov
         uses: codecov/codecov-action@v3
         with:
           directory: coverage
-
-  ci-notify:
-    runs-on: ubuntu-latest
-    if: ${{ always() }}
-    needs:
-      - ci-docs
-      - ci-unit
-      - ci-integration
-    steps:
-      - name: Inform the Codemonkeys
-        uses: 8398a7/action-slack@v3
-        continue-on-error: true
-        with:
-          status: custom
-          fields: workflow,job,commit,repo,ref,author,took
-          custom_payload: |
-            {
-              username: 'action-slack',
-              icon_emoji: ':octocat:',
-              attachments: [{
-                color: '${{ needs.ci-test.result }}' === 'success' ? 'good' : '${{ needs.ci-test.result }}' === 'failure' ? 'danger' : 'warning',
-                text: `${process.env.AS_REPO}@${process.env.AS_REF}\n ${process.env.AS_WORKFLOW} (${process.env.AS_COMMIT})\n by ${process.env.AS_AUTHOR}\n Status: ${{ needs.ci-test.result }}`,
-              }]
-            }
-        env:
-          GITHUB_TOKEN: ${{ github.token }} # required
-          SLACK_WEBHOOK_URL: ${{ secrets.SLACK_WEBHOOK_URL }} # required
-          MATRIX_CONTEXT: ${{ toJson(matrix) }} # required
+          fail_ci_if_error: true
+
+  # TODO(rousik): The following slack notification gives no value and
+  # needs to be fixed. Until then, it might be better to do nothing
+  # at all.
+  # ci-notify:
+  #   name: Notify slack
+  #   runs-on: ubuntu-latest
+  #   if: ${{ always() }}
+  #   needs:
+  #     - ci-unit
+  #     - ci-integration
+  #   steps:
+  #     - name: Inform the Codemonkeys
+  #       uses: 8398a7/action-slack@v3
+  #       continue-on-error: true
+  #       with:
+  #         status: custom
+  #         fields: workflow,job,commit,repo,ref,author,took
+  #         custom_payload: |
+  #           {
+  #             username: 'action-slack',
+  #             icon_emoji: ':octocat:',
+  #             attachments: [{
+  #               color: '${{ needs.ci-test.result }}' === 'success' ? 'good' : '${{ needs.ci-test.result }}' === 'failure' ? 'danger' : 'warning',
+  #               text: `${process.env.AS_REPO}@${process.env.AS_REF}\n ${process.env.AS_WORKFLOW} (${process.env.AS_COMMIT})\n by ${process.env.AS_AUTHOR}\n Status: ${{ needs.ci-test.result }}`,
+  #             }]
+  #           }
+  #       env:
+  #         GITHUB_TOKEN: ${{ github.token }} # required
+  #         SLACK_WEBHOOK_URL: ${{ secrets.SLACK_WEBHOOK_URL }} # required
+  #         MATRIX_CONTEXT: ${{ toJson(matrix) }} # required

src/pudl/extract/dbf.py

@@ -5,20 +5,22 @@
 import warnings
 import zipfile
 from collections import defaultdict
-from collections.abc import Iterator
+from collections.abc import Callable, Iterator
 from functools import lru_cache
 from pathlib import Path
 from typing import IO, Any, Protocol, Self
 
 import pandas as pd
 import sqlalchemy as sa
+from dagster import op
 from dbfread import DBF, FieldParser
 
 import pudl
 import pudl.logging_helpers
 from pudl.metadata.classes import DataSource
 from pudl.settings import FercToSqliteSettings, GenericDatasetSettings
 from pudl.workspace.datastore import Datastore
+from pudl.workspace.setup import PudlPaths
 
 logger = pudl.logging_helpers.get_logger(__name__)
 
@@ -465,6 +467,30 @@ def get_db_path(self) -> str:
         db_path = str(Path(self.output_path) / self.DATABASE_NAME)
         return f"sqlite:///{db_path}"
 
+    @classmethod
+    def get_dagster_op(cls) -> Callable:
+        """Returns dagstger op that runs this extractor."""
+
+        @op(
+            name=f"dbf_{cls.DATASET}",
+            required_resource_keys={
+                "ferc_to_sqlite_settings",
+                "datastore",
+                "runtime_settings",
+            },
+        )
+        def inner_method(context) -> None:
+            """Instantiates dbf extractor and runs it."""
+            dbf_extractor = cls(
+                datastore=context.resources.datastore,
+                settings=context.resources.ferc_to_sqlite_settings,
+                clobber=context.resources.runtime_settings.clobber,
+                output_path=PudlPaths().output_dir,
+            )
+            dbf_extractor.execute()
+
+        return inner_method
+
     def execute(self):
         """Runs the extraction of the data from dbf to sqlite."""
         logger.info(

src/pudl/extract/ferc.py

@@ -1,42 +1,17 @@
 """Hooks to integrate ferc to sqlite functionality into dagster graph."""
 
 
-from dagster import Field, op
-
 import pudl
 from pudl.extract.ferc1 import Ferc1DbfExtractor
 from pudl.extract.ferc2 import Ferc2DbfExtractor
 from pudl.extract.ferc6 import Ferc6DbfExtractor
 from pudl.extract.ferc60 import Ferc60DbfExtractor
-from pudl.workspace.setup import PudlPaths
 
 logger = pudl.logging_helpers.get_logger(__name__)
 
-
-@op(
-    config_schema={
-        "clobber": Field(
-            bool, description="Clobber existing ferc1 database.", default_value=False
-        ),
-    },
-    required_resource_keys={"ferc_to_sqlite_settings", "datastore"},
-)
-def dbf2sqlite(context) -> None:
-    """Clone the FERC Form 1 Visual FoxPro databases into SQLite."""
-    # TODO(rousik): this thin wrapper seems to be somewhat quirky. Maybe there's a way
-    # to make the integration # between the class and dagster better? Investigate.
-    logger.info(f"dbf2sqlite settings: {context.resources.ferc_to_sqlite_settings}")
-
-    extractors = [
-        Ferc1DbfExtractor,
-        Ferc2DbfExtractor,
-        Ferc6DbfExtractor,
-        Ferc60DbfExtractor,
-    ]
-    for xclass in extractors:
-        xclass(
-            datastore=context.resources.datastore,
-            settings=context.resources.ferc_to_sqlite_settings,
-            clobber=context.op_config["clobber"],
-            output_path=PudlPaths().output_dir,
-        ).execute()
+ALL_DBF_EXTRACTORS = [
+    Ferc1DbfExtractor,
+    Ferc2DbfExtractor,
+    Ferc6DbfExtractor,
+    Ferc60DbfExtractor,
+]