Enhanced Aircan with refactored codebase and v3 api support #40
Conversation
📝 WalkthroughWalkthroughRestructures the extension from Changes
Sequence Diagram(s)sequenceDiagram
participant User
participant Browser
participant CKAN_Plugin as CKAN Plugin
participant Actions as Action Logic
participant Airflow as Airflow API
User->>Browser: Click "Run pipeline" on resource page
Browser->>CKAN_Plugin: POST /dataset/<id>/resource_pipeline/<resource_id>
CKAN_Plugin->>Actions: aircan_submit(context, data_dict)
Actions->>CKAN_Plugin: plugin.update_payload(context, payload)
Actions->>Airflow: AirflowClient.trigger_dag(conf)
Airflow-->>Actions: DAG run response (dag_run_id / status)
Actions->>CKAN_Plugin: aircan_status_update(... dag_run_id ...)
CKAN_Plugin-->>Browser: Redirect / render pipeline page with status
Estimated code review effort🎯 5 (Critical) | ⏱️ ~120 minutes Poem
🚥 Pre-merge checks | ✅ 2 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (2 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. ✨ Finishing touches
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 14
🤖 Fix all issues with AI agents
In @.gitignore:
- Line 22: Remove the duplicate "__pycache__/" entry in the .gitignore (the
second occurrence around line 22) since it is already listed earlier in the
"Byte-compiled / optimized / DLL files" section; delete the redundant line and
scan the rest of .gitignore for any other duplicate entries to keep the file
concise.
In @ckanext/aircan/helpers.py:
- Around line 11-14: Remove the bypass of authorization by deleting the
"ignore_auth": True entry from the context dict used when calling resource_show
in the helper that builds the context for fetching pipeline status; ensure the
helper passes the normal context (without ignore_auth) so CKAN's resource_show
authorization is enforced (no other changes needed to template usage since
templates render only visible resources).
In @ckanext/aircan/lib/airflow.py:
- Around line 90-112: In request, handle expired v2 tokens by detecting 401
responses: after sending the request in the api_version == "v2" path, if
resp.status_code == 401 call self._login_local() to refresh self._access_token
and retry the request once (use the same headers with updated Authorization from
self._access_token), and propagate the final response; and for non-v2 basic auth
add explicit credential validation before using requests.auth.HTTPBasicAuth —
check that self.username and self.password are not None (raise a clear
RuntimeError mentioning username/password) so missing credentials fail fast
rather than producing obscure auth errors.
In @ckanext/aircan/logic/action.py:
- Around line 18-31: The function aircan_submit can return None on the
disallowed-format branch but its signature declares Dict[str, Any]; change the
signature to return Optional[Dict[str, Any]] (import typing.Optional) and ensure
all return paths are consistent with Optional[Dict[str, Any]] (or alternatively
replace the early return with an explicit empty dict return {} if you prefer to
keep Dict[str, Any]); update the type hint and any callers or mypy annotations
accordingly so the return type matches the actual behavior.
- Around line 44-64: The try/except around AirflowClient().trigger_dag swallows
requests.HTTPError and allows execution to continue, leaving dag_run undefined;
update the except block in the code that uses AirflowClient, trigger_dag and
dag_run to either re-raise the caught requests.HTTPError (or raise a new
exception) after logging, or return an error result so subsequent references to
dag_run (used by tk.get_action("aircan_status_update") and the returned dict)
are never executed when trigger_dag fails; ensure the change is made in the
block that constructs dag_run and calls tk.get_action("aircan_status_update") so
dag_run is always defined when used.
- Around line 118-119: The code raises p.toolkit.ValidationError but `p` is not
defined; fix by either importing the plugins alias or using the toolkit
directly—e.g., add "import ckan.plugins as p" at the top of the module so the
existing reference p.toolkit.ValidationError works, or replace
p.toolkit.ValidationError with toolkit.ValidationError and import the toolkit
(from ckan.plugins import toolkit) to ensure ValidationError is defined for the
resource_id check in the function containing that conditional.
In @ckanext/aircan/templates/resource_pipeline.html:
- Around line 86-92: The button element in the template has an extra
double-quote after the Jinja conditional `{% if state in ['running', 'queued',
'pending'] %}disabled aria-disabled="true"{% endif %}"` which produces malformed
HTML; edit the button tag around the conditional (the `<button ... {% if state
in ['running', 'queued', 'pending'] %}disabled aria-disabled="true"{% endif %}>`
line) and remove the stray trailing quote so the tag ends correctly without the
extra quote.
In @ckanext/aircan/tests/logic/test_validators.py:
- Line 10: Rename the misspelled test functions that use "reauired" to
"required": update test_aircan_reauired_with_valid_value to
test_aircan_required_with_valid_value and the other
test_aircan_reauired_with_invalid_value (line 14) to
test_aircan_required_with_invalid_value, and adjust any references or
decorators/imports that reference those function names so tests run correctly.
In @ckanext/aircan/tests/test_helpers.py:
- Around line 6-7: The test test_aircan_hello calls a non-existent function
helpers.aircan_hello(), causing an AttributeError; either remove the obsolete
test or update it to exercise existing helpers such as replacing the assertion
to call helpers.get_aircan_badge() (or use helpers.get_helpers() appropriately)
and assert the expected return value/structure, ensuring the test references
valid symbols (test_aircan_hello -> helpers.get_aircan_badge or
helpers.get_helpers) and validates their actual outputs.
In @ckanext/aircan/tests/test_plugin.py:
- Around line 50-56: The test is missing imports for the pytest decorators and
the plugin_loaded helper, so add an import for pytest and import plugin_loaded
(e.g., from ckan.tests.helpers import plugin_loaded) at the top of the file so
the pytest.mark decorators and the plugin_loaded("aircan") call in test_plugin()
are defined.
In @ckanext/aircan/tests/test_views.py:
- Around line 13-16: The test function test_aircan_blueprint references a
non-existent endpoint "aircan.page" and expects "Hello, aircan!" while the
codebase only registers the resource_pipeline route (endpoint name
"resource_pipeline") that renders resource_pipeline.html with status data;
update the test to call the existing endpoint (use
app.get(tk.h.url_for("resource_pipeline", id=<valid_dataset_id>,
resource_id=<valid_resource_id>)) or direct path
/dataset/<id>/resource_pipeline/<resource_id>), assert resp.status_code == 200
and check for expected content from resource_pipeline.html (or use
template/assertions that match the rendered status data), and remove the unused
reset_db parameter from the test signature; alternatively, if you prefer to keep
the original test behavior implement a new route/endpoint "aircan.page" in
views.py that returns "Hello, aircan!" and then keep the test but still remove
the unused reset_db param.
In @pyproject.toml:
- Line 13: Remove the Python 3.8 classifier entry from pyproject.toml by
deleting the "Programming Language :: Python :: 3.8" line (i.e., update the
classifiers list to exclude that exact string) so the project no longer
advertises support for EOL Python 3.8.
In @README.md:
- Line 4: Replace the misspelled "Aiflow" with "Airflow" in the README sentence
describing the extension; update the phrase "display the status and logs of
Aiflow data ingestion flows" to "display the status and logs of Airflow data
ingestion flows" so the documentation uses the correct product name.
- Around line 7-9: The Features section uses "Prefect" but the project targets
Airflow; update the two bullets under the "## Features" header to replace
"Prefect" with "Airflow" (e.g., change "Trigger Prefect Flows" to "Trigger
Airflow DAGs" and "Prefect flow runs" to "Airflow DAG runs"), and scan the
README for any other occurrences of "Prefect" and replace them with the
appropriate Airflow terminology to keep docs consistent.
🧹 Nitpick comments (19)
ckanext/aircan/tests/logic/test_validators.py (1)
10-16: Consider adding edge case tests.The current tests cover the basic scenarios, but you might want to add tests for edge cases such as:
- Empty string
""- Whitespace-only strings
- Other falsy values like
0,False,[],{}This would help ensure the validator behaves correctly across all input types.
ckanext/aircan/assets/style.css (1)
1-5: Consider removing this unused CSS file.This file contains only commented-out code and is not referenced by the webassets configuration (which uses
css/aircan.cssinstead). Keeping unused placeholder files can cause confusion for future maintainers.ckanext/aircan/logic/schema.py (1)
4-11: Consider renaming the function to better reflect its purpose.The function name
aircan_get_sum()suggests it returns a sum value, but it actually returns a validation schema. Consider renaming toaircan_get_sum_schema()oraircan_sum_schema()to better convey that it returns a dictionary of validators, not a computed sum.♻️ Proposed refactor
-def aircan_get_sum(): +def aircan_get_sum_schema(): not_empty = tk.get_validator("not_empty") convert_int = tk.get_validator("convert_int") return { "left": [not_empty, convert_int], "right": [not_empty, convert_int] }dev-requirements.txt (1)
1-1: Pin the pytest-ckan version in dev-requirements.txt.Currently unpinned dependencies can pull incompatible versions during future installations, breaking the test suite. The project already uses version pinning for runtime dependencies (e.g.,
google-auth>=2.46.0in pyproject.toml); the same approach should be applied here. Pin to at leastpytest-ckan>=0.0.12or use a compatible release constraint likepytest-ckan~=0.0.12.pyproject.toml (1)
19-21: The minimum versiongoogle-auth>=2.46.0is compatible with Airflow v2/v3.Version 2.46.0 (released Jan 5, 2026) exceeds the current stable apache-airflow-providers-google requirement of >=2.29.0. The version introduced support for workload certificate config and mTLS for Agentic Identities. However, consider adding an upper bound (e.g.,
google-auth>=2.46.0,<3.0.0) to prevent breaking changes from future major versions—there are precedents for this in older Airflow provider releases, and the 1.x → 2.x transition demonstrated significant API changes.ckanext/aircan/assets/webassets.yml (1)
1-14: Entire webassets configuration is commented out.All asset bundle definitions are commented. If this is intentional scaffolding for future use, consider adding a comment explaining this. Otherwise, if assets should be active, uncomment the configuration.
Also note the paths reference
js/aircan.jsandcss/aircan.css, but the actual JavaScript file appears to be atscript.jsbased on the file structure.ckanext/aircan/utils.py (1)
1-1: Unused import.
jsonis imported but not used in this file. Consider removing it.♻️ Remove unused import
-import json - import ckan.plugins.toolkit as tkckanext/aircan/cli.py (1)
11-16: Consider explicitly settingrequired=Falsewhen providing a default argument.While
click.argumentdoes support thedefaultparameter, it's clearer to explicitly setrequired=Falsewhen providing a default value. This makes the intent explicit that the argument is optional:@aircan.command() @click.argument("name", default="aircan", required=False) def command(name): """Docs. """ click.echo("Hello, {name}!".format(name=name)).github/workflows/test.yml (1)
21-22: Update Redis image to a supported version.
redis:3reached end-of-life on July 31, 2023. Consider upgrading toredis:7, which has been stable since April 2022.♻️ Suggested update
redis: - image: redis:3 + image: redis:7README.md (2)
44-56: Add language specifier to fenced code block.The config settings code block should specify a language for better syntax highlighting and to satisfy markdown linting rules.
📝 Proposed fix
-``` +```ini ckanext.aircan.endpoint = http://localhost:8080
108-117: Add language specifier to code block.This code example should specify
pythonas the language.📝 Proposed fix
-``` +```python class ExamplePlugin(plugins.SingletonPlugin):ckanext/aircan/tests/logic/test_action.py (1)
8-13: LGTM, but consider expanding test coverage.The test correctly validates the
aircan_get_sumaction with proper CKAN test fixtures. Consider adding tests for the actual Airflow integration actions (aircan_submit,aircan_status,aircan_status_logs) documented in the README, even if they require mocking the Airflow client.ckanext/aircan/tests/test_views.py (1)
5-5: Remove unused import.The
validatorsmodule is imported but never used in this test file.♻️ Proposed fix
-import ckanext.aircan.validators as validators - - import ckan.plugins.toolkit as tkckanext/aircan/helpers.py (1)
36-43: Consider extracting the badge HTML to a template snippet.While the inline HTML generation works, consider moving this to a Jinja2 template snippet (e.g.,
badge.html) for better maintainability and separation of concerns. This would make it easier to modify the badge styling without touching Python code.ckanext/aircan/logic/auth.py (1)
17-20: Minor type hint inconsistency.The
data_dictparameter is missing theDataDicttype annotation, unlike the other auth functions.♻️ Proposed fix
-def aircan_status_update(context: Context, data_dict: DataDict) -> AuthResult: +def aircan_status_update(context: Context, data_dict: DataDict) -> AuthResult: return authz.is_authorized( "resource_update", context, {"id": data_dict.get("resource_id")} )Wait, looking again at the provided code, the type hint is already there. Let me re-read...
Actually looking at line 17:
def aircan_status_update(context: Context, data_dict: DataDict) -> AuthResult:- this is correct. I misread. Disregard.ckanext/aircan/views.py (2)
41-46: Remove redundantpassstatement.The
passstatement on line 42 is dead code since it's immediately followed by another statement.♻️ Proposed fix
except logic.ValidationError: - pass tk.h.flash_error( tk._("There was an error submitting the resource for processing.") )
47-52: Consider using endpoint-style redirect for consistency with Flask blueprints.While the
controller="aircan"syntax is valid and supported by CKAN'stoolkit.redirect_to()wrapper, using the endpoint-style approach ('aircan.resource_pipeline') would be more consistent with modern Flask blueprint conventions. Both syntaxes work, but endpoint-style is preferred for new code.ckanext/aircan/plugin.py (1)
31-53: Consider checking conditions before fetching resource.The
notifymethod fetches the full resource viaresource_showbefore checking ifurl_changedandlast_modifiedare set. Moving the condition check before the API call would avoid unnecessary database queries.♻️ Proposed optimization
def notify(self, entity, operation): """ Notify the plugin of a domain object modification. """ if operation != DomainObjectOperation.changed or not isinstance( entity, Resource ): return + if not getattr(entity, "url_changed", False) or not getattr( + entity, "last_modified", False + ): + return + context = { "ignore_auth": True, } resource_dict = tk.get_action("resource_show")( context, { "id": entity.id, }, ) - if not getattr(entity, "url_changed", False) or not getattr( - entity, "last_modified", False - ): - return self._self_aircan_submit(resource_dict)ckanext/aircan/logic/action.py (1)
83-97: Handle potential JSON parsing error and uselog.exception.Line 84 parses JSON without error handling. If
valueis malformed, this will raise an exception. Also, usinglog.exceptioninstead oflog.errorwill include the stack trace.♻️ Proposed fix
if task_status: - dag_run_id = json.loads(task_status.get("value", "{}")).get("dag_run_id", "") + try: + dag_run_id = json.loads(task_status.get("value", "{}")).get("dag_run_id", "") + except json.JSONDecodeError: + dag_run_id = "" client = AirflowClient() try: dag_run = client.get_dag_run(dag_run_id) task_status.update( { "dag": dag_run, } ) except requests.HTTPError as e: - log.error( + log.exception( tk._("Failed to fetch Airflow DAG run '%s': %s"), dag_run_id, str(e) )
📜 Review details
Configuration used: defaults
Review profile: CHILL
Plan: Pro
⛔ Files ignored due to path filters (6)
cypress/fixtures/sample-pdf-large-size.pdfis excluded by!**/*.pdfcypress/fixtures/sample-pdf-with-images.pdfis excluded by!**/*.pdfcypress/fixtures/sample.csvis excluded by!**/*.csvcypress/fixtures/sample.xlsxis excluded by!**/*.xlsxcypress/fixtures/sample2.csvis excluded by!**/*.csvpackage-lock.jsonis excluded by!**/package-lock.json
📒 Files selected for processing (72)
.coveragerc.github/workflows/test.yml.gitignore.travis.ymlLICENSEMANIFEST.inREADME.mdREADME.rstbin/travis-build.bashbin/travis-run.shckanext/aircan/__init__.pyckanext/aircan/assets/.gitignoreckanext/aircan/assets/script.jsckanext/aircan/assets/style.cssckanext/aircan/assets/webassets.ymlckanext/aircan/cli.pyckanext/aircan/helpers.pyckanext/aircan/i18n/.gitignoreckanext/aircan/interfaces.pyckanext/aircan/lib/airflow.pyckanext/aircan/logic/action.pyckanext/aircan/logic/auth.pyckanext/aircan/logic/schema.pyckanext/aircan/plugin.pyckanext/aircan/public/.gitignoreckanext/aircan/templates/package/resource_edit_base.htmlckanext/aircan/templates/package/snippets/resource_info.htmlckanext/aircan/templates/package/snippets/resource_item.htmlckanext/aircan/templates/resource_pipeline.htmlckanext/aircan/tests/__init__.pyckanext/aircan/tests/logic/test_action.pyckanext/aircan/tests/logic/test_auth.pyckanext/aircan/tests/logic/test_validators.pyckanext/aircan/tests/test_helpers.pyckanext/aircan/tests/test_plugin.pyckanext/aircan/tests/test_views.pyckanext/aircan/utils.pyckanext/aircan/views.pyckanext/aircan_connector/blueprint.pyckanext/aircan_connector/fanstatic/css/aircan.cssckanext/aircan_connector/fanstatic/js/aircan.jsckanext/aircan_connector/fanstatic/resource.configckanext/aircan_connector/logic/action.pyckanext/aircan_connector/logic/auth.pyckanext/aircan_connector/logic/dag_status_report.pyckanext/aircan_connector/logic/gcp_handler.pyckanext/aircan_connector/logic/helpers.pyckanext/aircan_connector/plugin.pyckanext/aircan_connector/templates/.gitignoreckanext/aircan_connector/templates/package/resource_edit_base.htmlckanext/aircan_connector/templates/resource_data.htmlckanext/aircan_connector/tests/__init__.pyckanext/aircan_connector/tests/test_plugin.pyckanext/aircan_connector/tests/test_submit_action_local_airflow.pycypress.jsoncypress/fixtures/aircan.jsoncypress/integration/aircan-connector.jscypress/integration/ckan-classic-api.jscypress/integration/ckan-classic-auth.jscypress/integration/ckan-classic-ui.jscypress/integration/ckan-datastore.jscypress/plugins/index.jscypress/support/commands.jscypress/support/index.jsdev-requirements.txtpackage.jsonpyproject.tomlrequirements.txtruntest.shsetup.cfgsetup.pytest.ini
💤 Files with no reviewable changes (31)
- bin/travis-run.sh
- ckanext/aircan_connector/fanstatic/resource.config
- cypress/support/index.js
- ckanext/aircan_connector/tests/test_submit_action_local_airflow.py
- requirements.txt
- cypress/plugins/index.js
- .travis.yml
- cypress/integration/ckan-classic-api.js
- cypress/integration/ckan-classic-auth.js
- cypress/integration/ckan-classic-ui.js
- runtest.sh
- ckanext/aircan_connector/logic/helpers.py
- cypress/support/commands.js
- cypress.json
- ckanext/aircan_connector/logic/dag_status_report.py
- bin/travis-build.bash
- ckanext/aircan_connector/logic/gcp_handler.py
- ckanext/aircan_connector/logic/auth.py
- ckanext/aircan_connector/templates/package/resource_edit_base.html
- README.rst
- ckanext/aircan_connector/fanstatic/js/aircan.js
- cypress/fixtures/aircan.json
- cypress/integration/aircan-connector.js
- ckanext/aircan_connector/plugin.py
- cypress/integration/ckan-datastore.js
- package.json
- ckanext/aircan_connector/logic/action.py
- ckanext/aircan_connector/tests/test_plugin.py
- ckanext/aircan_connector/templates/resource_data.html
- ckanext/aircan_connector/fanstatic/css/aircan.css
- ckanext/aircan_connector/blueprint.py
🧰 Additional context used
🧬 Code graph analysis (9)
ckanext/aircan/views.py (4)
ckanext/aircan/lib/airflow.py (1)
request(90-112)ckanext/aircan/cli.py (1)
aircan(5-8)ckanext/aircan/logic/action.py (1)
aircan_status(67-99)ckanext/aircan/logic/auth.py (1)
aircan_status(11-14)
ckanext/aircan/tests/logic/test_auth.py (1)
ckanext/aircan/tests/logic/test_action.py (1)
test_aircan_get_sum(10-13)
ckanext/aircan/logic/auth.py (2)
ckanext/aircan/logic/action.py (2)
aircan_status(67-99)aircan_status_update(102-170)ckanext/aircan/plugin.py (1)
get_auth_functions(63-64)
ckanext/aircan/lib/airflow.py (1)
ckanext/aircan/views.py (1)
get(54-94)
ckanext/aircan/utils.py (1)
ckanext/aircan/views.py (1)
get(54-94)
ckanext/aircan/logic/action.py (4)
ckanext/aircan/utils.py (1)
allowed_formats(17-28)ckanext/aircan/lib/airflow.py (3)
AirflowClient(16-153)trigger_dag(114-146)get_dag_run(148-153)ckanext/aircan/interfaces.py (2)
IAircan(4-14)update_payload(7-14)ckanext/aircan/plugin.py (2)
update_payload(79-86)get_actions(67-68)
ckanext/aircan/tests/logic/test_action.py (1)
ckanext/aircan/tests/logic/test_auth.py (1)
test_aircan_get_sum(12-19)
ckanext/aircan/tests/test_views.py (1)
ckanext/aircan/views.py (1)
get(54-94)
ckanext/aircan/helpers.py (4)
ckanext/aircan/logic/action.py (1)
aircan_status(67-99)ckanext/aircan/logic/auth.py (1)
aircan_status(11-14)ckanext/aircan/views.py (1)
get(54-94)ckanext/aircan/plugin.py (1)
get_helpers(71-72)
🪛 Checkov (3.2.334)
.github/workflows/test.yml
[medium] 25-26: Basic Auth Credentials
(CKV_SECRET_4)
🪛 LanguageTool
README.md
[grammar] ~4-~4: Ensure spelling is correct
Context: ...tor, and display the status and logs of Aiflow data ingestion flows directly from the ...
(QB_NEW_EN_ORTHOGRAPHY_ERROR_IDS_1)
🪛 markdownlint-cli2 (0.18.1)
README.md
30-30: Bare URL used
(MD034, no-bare-urls)
44-44: Fenced code blocks should have a language specified
(MD040, fenced-code-language)
62-62: Code block style
Expected: fenced; Actual: indented
(MD046, code-block-style)
108-108: Fenced code blocks should have a language specified
(MD040, fenced-code-language)
🪛 Ruff (0.14.10)
ckanext/aircan/views.py
15-15: Unused method argument: id
(ARG002)
15-15: Unused method argument: resource_id
(ARG002)
77-77: Do not catch blind exception: Exception
(BLE001)
84-84: Do not catch blind exception: Exception
(BLE001)
ckanext/aircan/plugin.py
79-79: Unused method argument: context
(ARG002)
ckanext/aircan/lib/airflow.py
50-50: Avoid specifying long messages outside the exception class
(TRY003)
54-56: Avoid specifying long messages outside the exception class
(TRY003)
70-72: Avoid specifying long messages outside the exception class
(TRY003)
86-86: Avoid specifying long messages outside the exception class
(TRY003)
98-98: Avoid specifying long messages outside the exception class
(TRY003)
110-110: Probable use of requests call without timeout
(S113)
140-142: Avoid specifying long messages outside the exception class
(TRY003)
ckanext/aircan/logic/action.py
48-48: Use logging.exception instead of logging.error
Replace with exception
(TRY400)
95-97: Use logging.exception instead of logging.error
Replace with exception
(TRY400)
119-119: Undefined name p
(F821)
ckanext/aircan/tests/test_plugin.py
53-53: Undefined name pytest
(F821)
54-54: Undefined name pytest
(F821)
56-56: Undefined name plugin_loaded
(F821)
ckanext/aircan/tests/test_views.py
13-13: Unused function argument: reset_db
(ARG001)
🔇 Additional comments (27)
.coveragerc (1)
1-5: LGTM!This is a housekeeping change that preserves the existing coverage configuration while likely adding a proper trailing newline. The omit patterns correctly exclude external dependencies and CKAN core from coverage reports.
LICENSE (1)
661-661: Verify the summary description of changes.The AI summary indicates that the period after the URL reference was removed, but Line 661 still shows
<http://www.gnu.org/licenses/>.with the period intact. This appears inconsistent with the summary description. Please clarify what the actual textual change is..gitignore (1)
42-42: Good hygiene improvement.The newline normalization on the
docs/_build/entry ensures consistent file formatting.ckanext/aircan/templates/package/resource_edit_base.html (1)
1-5: The routeaircan.resource_pipelineis properly registered and the template is correctly implemented.The route is registered in
ckanext/aircan/views.py(lines 97-100) with Blueprint "aircan" and endpoint "resource_pipeline", which correctly resolves to the full endpoint name "aircan.resource_pipeline" used in the template. The block override ofinner_primary_navis an intentional design choice to provide custom resource pipeline navigation in the resource edit interface, consistent with other template customizations in the extension.Likely an incorrect or invalid review comment.
MANIFEST.in (1)
4-6: LGTM!The manifest updates correctly reflect the restructuring from
aircan_connectortoaircan, and properly include all necessary asset types, migrations, and public files for the CKAN extension.pyproject.toml (1)
34-40: LGTM! Pytest configuration is appropriate.The pytest configuration correctly filters SQLAlchemy deprecation warnings and points to the CKAN test configuration file.
test.ini (1)
1-45: LGTM!The test configuration changes are appropriate:
- The updated
error_email_fromaligns with CKAN conventions.- The relative path adjustment (
../ckan/test-core.ini) is consistent with the extension's new directory structure.- Logging to
stderris a reasonable choice, keepingstdoutclean for test output.ckanext/aircan/cli.py (1)
19-20: LGTM!The
get_commands()function follows the standard CKAN pattern for exposing CLI commands to the plugin system.ckanext/aircan/interfaces.py (1)
1-14: LGTM!The interface follows standard CKAN plugin patterns. The
update_payloadmethod provides a clear extension point for plugins to modify payloads before Airflow submission..github/workflows/test.yml (2)
24-29: Test database credentials are acceptable for CI.The static analysis hint flagged hardcoded credentials in the database URLs. These are standard CKAN test environment credentials for ephemeral CI containers and pose no security risk. No action needed.
31-47: LGTM!The workflow steps follow standard CKAN extension testing patterns: checkout, install dependencies, configure test.ini for container paths, initialize database, and run pytest with coverage.
ckanext/aircan/assets/script.js (1)
1-10: LGTM!This is a minimal CKAN JavaScript module scaffold following the standard pattern. The empty
initializefunction can be extended as needed for client-side functionality.ckanext/aircan/utils.py (1)
17-28: LGTM!The
allowed_formatsfunction correctly handles:
- Configuration override via
ckanext.aircan.formats- Case-insensitive format matching
- Falsy input guard
setup.cfg (1)
1-40: LGTM!The Babel/i18n configuration has been properly updated to reflect the package rename from
aircan_connectortoaircan. All domain names and file paths are consistently aligned acrossextract_messages,init_catalog,update_catalog, andcompile_catalogsections.setup.py (1)
1-21: LGTM!The simplified
setup.pyfollows modern Python packaging practices by moving metadata topyproject.tomlwhile retaining only Babel message extraction configuration. The reference to CKAN issue #8382 provides helpful context for this pattern.ckanext/aircan/lib/airflow.py (3)
68-88: LGTM!The
_login_localmethod properly validates credentials before attempting authentication and correctly handles the token response.
148-153: LGTM!The
get_dag_runmethod correctly fetches DAG run details and propagates errors viaraise_for_status().
134-146: Unreachable code afterraise_for_status().Lines 144-146 are unreachable:
- 200/201 returns early at line 137
- 403 raises at line 140-142
- Any other non-2xx status will cause
raise_for_status()at line 143 to raise an exception🧹 Proposed fix
if resp.status_code in (200, 201): data = resp.json() data.setdefault("dag_run_id", dag_run_id) return data if resp.status_code == 403: raise requests.HTTPError( "Forbidden: Check Airflow RBAC roles/permissions for this operation." ) resp.raise_for_status() - data = resp.json() - data.setdefault("dag_run_id", dag_run_id) - return data + # Unreachable: raise_for_status() raises for non-2xx responses + return {} # type: ignore[unreachable]Or simply remove lines 144-146 entirely.
Likely an incorrect or invalid review comment.
ckanext/aircan/templates/package/snippets/resource_info.html (1)
1-6: LGTM!The template correctly extends the base CKAN template and appends the Aircan badge to the resource info block. Using
| safeis appropriate here sinceh.get_aircan_badge()generates trusted HTML.ckanext/aircan/templates/package/snippets/resource_item.html (1)
1-6: LGTM!The template follows the same pattern as
resource_info.html, consistently extending the base template and appending the Aircan badge to the resource item title.ckanext/aircan/tests/logic/test_auth.py (1)
10-19: LGTM!The authorization test is well-structured with proper fixtures, context setup, and assertion logic.
ckanext/aircan/templates/resource_pipeline.html (2)
109-117: LGTM!The error display section correctly checks for
status.errorbefore rendering. The view layer ensures the error object is normalized to contain amessagekey.
147-205: LGTM!The logs section handles empty states gracefully, uses appropriate accessibility attributes, and the
urlizefilter safely converts URLs in log messages to clickable links.ckanext/aircan/logic/auth.py (1)
1-28: LGTM!The authorization functions correctly delegate to CKAN's built-in authorization system, mapping each Aircan action to the appropriate resource permission (
resource_create,resource_show,resource_update).ckanext/aircan/views.py (1)
71-86: LGTM!The status normalization logic defensively handles JSON parsing failures and ensures the error object always has a consistent structure for the template.
ckanext/aircan/plugin.py (1)
78-86: LGTM!The
update_payloadmethod correctly implements theIAircaninterface. Thecontextparameter is part of the interface contract and is intentionally passed through even if not used in the base implementation, allowing subclasses to utilize it.ckanext/aircan/logic/action.py (1)
173-178: LGTM!The
get_actionsfunction correctly exposes all three action functions for registration with CKAN's plugin system.
| .installed.cfg | ||
| *.egg | ||
|
|
||
| __pycache__/ |
There was a problem hiding this comment.
Remove duplicate __pycache__/ entry.
Line 6 already includes __pycache__/ in the "Byte-compiled / optimized / DLL files" section. The duplicate entry on line 22 is redundant and should be removed to maintain clarity.
🧹 Proposed fix
*.spec
# Installer logs
pip-log.txt
pip-delete-this-directory.txt
# Unit test / coverage reports
htmlcov/
.tox/
.coverage
.cache
nosetests.xml
coverage.xml
# Sphinx documentation
-__pycache__/
docs/_build/Committable suggestion skipped: line range outside the PR's diff.
🤖 Prompt for AI Agents
In @.gitignore at line 22, Remove the duplicate "__pycache__/" entry in the
.gitignore (the second occurrence around line 22) since it is already listed
earlier in the "Byte-compiled / optimized / DLL files" section; delete the
redundant line and scan the rest of .gitignore for any other duplicate entries
to keep the file concise.
There was a problem hiding this comment.
@copilot In @.gitignore at line 22, Remove the duplicate "pycache/" entry in the
.gitignore (the second occurrence around line 22) since it is already listed
earlier in the "Byte-compiled / optimized / DLL files" section; delete the
redundant line and scan the rest of .gitignore for any other duplicate entries
to keep the file concise.
There was a problem hiding this comment.
Seems like the humans are having a chat. I'll hop back into my burrow for now. If you need me again, just tag @coderabbitai in a new comment, and I'll come hopping out!
| context = { | ||
| "model": model, | ||
| "ignore_auth": True, | ||
| } |
There was a problem hiding this comment.
🧩 Analysis chain
🏁 Script executed:
# First, let's read the helpers.py file to understand the full context
cd ckanext/aircan
wc -l helpers.pyRepository: datopian/ckanext-aircan
Length of output: 80
🏁 Script executed:
# Read the helpers.py file with context around the flagged lines
cat -n helpers.pyRepository: datopian/ckanext-aircan
Length of output: 109
🏁 Script executed:
# Search for ignore_auth usage in the codebase to understand the pattern
rg -n "ignore_auth.*True" --type=py -B 2 -A 2Repository: datopian/ckanext-aircan
Length of output: 888
🏁 Script executed:
# Read the helpers.py file with correct path
cat -n ckanext/aircan/helpers.pyRepository: datopian/ckanext-aircan
Length of output: 1838
🏁 Script executed:
# Also read plugin.py to see the pattern usage
cat -n ckanext/aircan/plugin.py | head -70Repository: datopian/ckanext-aircan
Length of output: 2663
🏁 Script executed:
# Search for how ignore_auth is used in CKAN core documentation
rg -n "ignore_auth" --type=py ckanext/aircan/Repository: datopian/ckanext-aircan
Length of output: 275
🏁 Script executed:
# Search for where get_aircan_badge is used in templates
rg -r "get_aircan_badge" --type=html --type=jinja2Repository: datopian/ckanext-aircan
Length of output: 463
🏁 Script executed:
# Also search in .j2 files and other template formats
find . -type f \( -name "*.html" -o -name "*.jinja2" -o -name "*.j2" \) -exec grep -l "get_aircan_badge" {} \;Repository: datopian/ckanext-aircan
Length of output: 192
🏁 Script executed:
# Check if there's any documentation or comments about authorization in the codebase
rg -n "ignore_auth" --type=py -B 5 -A 5 ckanext/aircan/logic/Repository: datopian/ckanext-aircan
Length of output: 49
🏁 Script executed:
# Check the logic directory structure
ls -la ckanext/aircan/logic/Repository: datopian/ckanext-aircan
Length of output: 353
🏁 Script executed:
# Check auth.py for authorization implementation
cat -n ckanext/aircan/logic/auth.pyRepository: datopian/ckanext-aircan
Length of output: 1083
🏁 Script executed:
# Check the templates where get_aircan_badge is used
cat ckanext/aircan/templates/package/snippets/resource_info.htmlRepository: datopian/ckanext-aircan
Length of output: 185
🏁 Script executed:
# Also check resource_item.html
cat ckanext/aircan/templates/package/snippets/resource_item.htmlRepository: datopian/ckanext-aircan
Length of output: 191
🏁 Script executed:
# Check if there are tests that show the expected behavior
find . -name "*test*.py" -type f | head -20Repository: datopian/ckanext-aircan
Length of output: 314
🏁 Script executed:
# Look for test files related to helpers
find . -path "*/tests/*" -name "*helper*" -o -path "*/test*" -name "*helper*"Repository: datopian/ckanext-aircan
Length of output: 105
🏁 Script executed:
# Check for any documentation or comments about the authorization strategy
rg "ignore_auth" -B 3 -A 3 --type=pyRepository: datopian/ckanext-aircan
Length of output: 1085
🏁 Script executed:
# Check the test_helpers.py to understand the intended behavior
cat -n ckanext/aircan/tests/test_helpers.pyRepository: datopian/ckanext-aircan
Length of output: 266
🏁 Script executed:
# Also check test_auth.py to see how authorization is tested
cat -n ckanext/aircan/tests/logic/test_auth.pyRepository: datopian/ckanext-aircan
Length of output: 664
🏁 Script executed:
# Check the action.py to see how aircan_status is implemented
cat -n ckanext/aircan/logic/action.py | head -100Repository: datopian/ckanext-aircan
Length of output: 3780
Remove ignore_auth: True to respect resource authorization.
The helper bypasses authorization checks by setting ignore_auth: True, which means any user can retrieve pipeline status for any resource regardless of visibility permissions. While the badge is currently used only in resource templates (where visibility is already controlled), the helper function itself can be called with arbitrary resource IDs, exposing internal pipeline information for private resources.
The aircan_status action (lines 11-14 of auth.py) correctly implements authorization through resource_show permissions. Removing ignore_auth: True allows this authorization to take effect without impacting template rendering, since the resource is already visible in that context.
🤖 Prompt for AI Agents
In @ckanext/aircan/helpers.py around lines 11 - 14, Remove the bypass of
authorization by deleting the "ignore_auth": True entry from the context dict
used when calling resource_show in the helper that builds the context for
fetching pipeline status; ensure the helper passes the normal context (without
ignore_auth) so CKAN's resource_show authorization is enforced (no other changes
needed to template usage since templates render only visible resources).
| def request(self, method: str, endpoint: str, **kwargs: Any) -> requests.Response: | ||
| if "timeout" not in kwargs: | ||
| kwargs["timeout"] = self.timeout | ||
|
|
||
| url = self._join_url(endpoint) | ||
|
|
||
| if self.server_type == "gcp": | ||
| if not self._authed_session: | ||
| raise RuntimeError("GCP session not initialized") | ||
| return self._authed_session.request(method, url, **kwargs) | ||
|
|
||
| headers = kwargs.pop("headers", {}) or {} | ||
|
|
||
| if self.api_version == "v2": | ||
| if not self._access_token: | ||
| self._login_local() | ||
| headers["Authorization"] = f"Bearer {self._access_token}" | ||
| kwargs["headers"] = headers | ||
| else: | ||
| kwargs["auth"] = requests.auth.HTTPBasicAuth(self.username, self.password) | ||
| resp = requests.request(method, url, **kwargs) | ||
|
|
||
| return resp |
There was a problem hiding this comment.
Missing token refresh and credential validation for non-v2 auth.
Two issues in the request method:
-
Token expiration: If the v2 JWT token expires mid-session, requests will fail with 401. Consider adding retry logic or token refresh on authentication failures.
-
Missing credential validation for non-v2: For
api_version != "v2", the code usesHTTPBasicAuthwithself.usernameandself.password, but these could beNoneif not configured, leading to authentication failures without a clear error message.
🔧 Proposed fix for credential validation
if self.api_version == "v2":
if not self._access_token:
self._login_local()
headers["Authorization"] = f"Bearer {self._access_token}"
kwargs["headers"] = headers
else:
+ if not self.username or not self.password:
+ raise ValueError(
+ "Missing config: ckanext.aircan.airflow_username / ckanext.aircan.airflow_password"
+ )
kwargs["auth"] = requests.auth.HTTPBasicAuth(self.username, self.password)
resp = requests.request(method, url, **kwargs)🧰 Tools
🪛 Ruff (0.14.10)
98-98: Avoid specifying long messages outside the exception class
(TRY003)
110-110: Probable use of requests call without timeout
(S113)
🤖 Prompt for AI Agents
In @ckanext/aircan/lib/airflow.py around lines 90 - 112, In request, handle
expired v2 tokens by detecting 401 responses: after sending the request in the
api_version == "v2" path, if resp.status_code == 401 call self._login_local() to
refresh self._access_token and retry the request once (use the same headers with
updated Authorization from self._access_token), and propagate the final
response; and for non-v2 basic auth add explicit credential validation before
using requests.auth.HTTPBasicAuth — check that self.username and self.password
are not None (raise a clear RuntimeError mentioning username/password) so
missing credentials fail fast rather than producing obscure auth errors.
ckanext/aircan/logic/action.py
Outdated
| client = AirflowClient() | ||
| try: | ||
| dag_run = client.trigger_dag(conf=payload) | ||
| except requests.HTTPError as e: | ||
| log.error(tk._("Failed to trigger Airflow DAG '%s': %s"), client.dag_id, str(e)) | ||
|
|
||
| context.update({"session": context["model"].meta.create_local_session()}) | ||
| tk.get_action("aircan_status_update")( | ||
| context, | ||
| { | ||
| "resource_id": data_dict.get("id"), | ||
| "dag_run_id": dag_run.get("dag_run_id"), | ||
| "state": "queued", | ||
| "message": f"Added to the queue to be processed with '{dag_run.get("dag_run_id")}'.", | ||
| "clear_logs": True, | ||
| }, | ||
| ) | ||
| return { | ||
| "dag_run": dag_run, | ||
| "dag_run_id": dag_run.get("dag_run_id"), | ||
| } |
There was a problem hiding this comment.
Critical: dag_run is undefined if trigger_dag raises HTTPError.
If client.trigger_dag() raises an HTTPError, the exception is logged but execution continues. The subsequent code on lines 55, 57, 62, and 63 references dag_run, which would be undefined, causing an UnboundLocalError.
🐛 Proposed fix - re-raise the exception
client = AirflowClient()
try:
dag_run = client.trigger_dag(conf=payload)
except requests.HTTPError as e:
- log.error(tk._("Failed to trigger Airflow DAG '%s': %s"), client.dag_id, str(e))
+ log.exception(tk._("Failed to trigger Airflow DAG '%s': %s"), client.dag_id, str(e))
+ raise tk.ValidationError(
+ {"airflow": [f"Failed to trigger Airflow DAG: {str(e)}"]}
+ )
context.update({"session": context["model"].meta.create_local_session()})🧰 Tools
🪛 Ruff (0.14.10)
48-48: Use logging.exception instead of logging.error
Replace with exception
(TRY400)
🤖 Prompt for AI Agents
In @ckanext/aircan/logic/action.py around lines 44 - 64, The try/except around
AirflowClient().trigger_dag swallows requests.HTTPError and allows execution to
continue, leaving dag_run undefined; update the except block in the code that
uses AirflowClient, trigger_dag and dag_run to either re-raise the caught
requests.HTTPError (or raise a new exception) after logging, or return an error
result so subsequent references to dag_run (used by
tk.get_action("aircan_status_update") and the returned dict) are never executed
when trigger_dag fails; ensure the change is made in the block that constructs
dag_run and calls tk.get_action("aircan_status_update") so dag_run is always
defined when used.
| <button class="btn btn-primary" | ||
| name="save" | ||
| type="submit" | ||
| {% if state in ['running', 'queued', 'pending'] %}disabled aria-disabled="true"{% endif %}"> | ||
| <i class="fa fa-diagram-project me-1"></i> | ||
| {{ _('Run Data Pipeline') }} | ||
| </button> |
There was a problem hiding this comment.
HTML syntax error: extra quote character.
Line 89 has an extra closing quote after the {% endif %} which will result in malformed HTML.
🔧 Proposed fix
<button class="btn btn-primary"
name="save"
type="submit"
- {% if state in ['running', 'queued', 'pending'] %}disabled aria-disabled="true"{% endif %}">
+ {% if state in ['running', 'queued', 'pending'] %}disabled aria-disabled="true"{% endif %}>
<i class="fa fa-diagram-project me-1"></i>
{{ _('Run Data Pipeline') }}
</button>📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| <button class="btn btn-primary" | |
| name="save" | |
| type="submit" | |
| {% if state in ['running', 'queued', 'pending'] %}disabled aria-disabled="true"{% endif %}"> | |
| <i class="fa fa-diagram-project me-1"></i> | |
| {{ _('Run Data Pipeline') }} | |
| </button> | |
| <button class="btn btn-primary" | |
| name="save" | |
| type="submit" | |
| {% if state in ['running', 'queued', 'pending'] %}disabled aria-disabled="true"{% endif %}> | |
| <i class="fa fa-diagram-project me-1"></i> | |
| {{ _('Run Data Pipeline') }} | |
| </button> |
🤖 Prompt for AI Agents
In @ckanext/aircan/templates/resource_pipeline.html around lines 86 - 92, The
button element in the template has an extra double-quote after the Jinja
conditional `{% if state in ['running', 'queued', 'pending'] %}disabled
aria-disabled="true"{% endif %}"` which produces malformed HTML; edit the button
tag around the conditional (the `<button ... {% if state in ['running',
'queued', 'pending'] %}disabled aria-disabled="true"{% endif %}>` line) and
remove the stray trailing quote so the tag ends correctly without the extra
quote.
| import ckanext.aircan.plugin as plugin | ||
|
|
||
|
|
||
| @pytest.mark.ckan_config("ckan.plugins", "aircan") | ||
| @pytest.mark.usefixtures("with_plugins") | ||
| def test_plugin(): | ||
| assert plugin_loaded("aircan") |
There was a problem hiding this comment.
Add missing imports.
The test uses pytest decorators and the plugin_loaded function without importing them, which will cause NameError at runtime.
🔧 Required fix
+import pytest
+
+from ckan.plugins import plugin_loaded
+
import ckanext.aircan.plugin as plugin📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| import ckanext.aircan.plugin as plugin | |
| @pytest.mark.ckan_config("ckan.plugins", "aircan") | |
| @pytest.mark.usefixtures("with_plugins") | |
| def test_plugin(): | |
| assert plugin_loaded("aircan") | |
| import pytest | |
| from ckan.plugins import plugin_loaded | |
| import ckanext.aircan.plugin as plugin | |
| @pytest.mark.ckan_config("ckan.plugins", "aircan") | |
| @pytest.mark.usefixtures("with_plugins") | |
| def test_plugin(): | |
| assert plugin_loaded("aircan") |
🧰 Tools
🪛 Ruff (0.14.10)
53-53: Undefined name pytest
(F821)
54-54: Undefined name pytest
(F821)
56-56: Undefined name plugin_loaded
(F821)
🤖 Prompt for AI Agents
In @ckanext/aircan/tests/test_plugin.py around lines 50 - 56, The test is
missing imports for the pytest decorators and the plugin_loaded helper, so add
an import for pytest and import plugin_loaded (e.g., from ckan.tests.helpers
import plugin_loaded) at the top of the file so the pytest.mark decorators and
the plugin_loaded("aircan") call in test_plugin() are defined.
| def test_aircan_blueprint(app, reset_db): | ||
| resp = app.get(tk.h.url_for("aircan.page")) | ||
| assert resp.status_code == 200 | ||
| assert resp.body == "Hello, aircan!" |
There was a problem hiding this comment.
🧩 Analysis chain
🏁 Script executed:
# First, locate the relevant files
find . -name "views.py" -o -name "test_views.py" | grep aircanRepository: datopian/ckanext-aircan
Length of output: 129
🏁 Script executed:
# Search for the aircan.page route definition
rg -n "aircan\.page|aircan_page" --type py -A 3 -B 3Repository: datopian/ckanext-aircan
Length of output: 552
🏁 Script executed:
# Get the full test file to understand context
find . -path "*/ckanext/aircan/tests/test_views.py" -exec cat {} \;Repository: datopian/ckanext-aircan
Length of output: 451
🏁 Script executed:
# Get the views.py file content
find . -path "*/ckanext/aircan/views.py" -exec wc -l {} \;Repository: datopian/ckanext-aircan
Length of output: 96
🏁 Script executed:
# Read the views.py file content
cat -n ./ckanext/aircan/views.pyRepository: datopian/ckanext-aircan
Length of output: 3779
🏁 Script executed:
# Also search for the blueprint definition and aircan.page route specifically
rg -n "Blueprint|aircan\.page|route.*page" ./ckanext/aircan/views.pyRepository: datopian/ckanext-aircan
Length of output: 138
🏁 Script executed:
# Search for any other route definitions, including the aircan.page route
rg -n "aircan\.page|\.route|add_url_rule" --type py -A 2 -B 2Repository: datopian/ckanext-aircan
Length of output: 753
🏁 Script executed:
# Also check if there are other Python files in the aircan extension
fd -e py . ckanext/aircan/Repository: datopian/ckanext-aircan
Length of output: 633
🏁 Script executed:
# Check the plugin.py file for any route definitions
cat -n ./ckanext/aircan/plugin.py | head -80Repository: datopian/ckanext-aircan
Length of output: 2997
🏁 Script executed:
# Search across the entire codebase for any definition of aircan.page route
rg -r "aircan.page|\.page" --type py -B 3 -A 3Repository: datopian/ckanext-aircan
Length of output: 463
🏁 Script executed:
# Search for "page" route definition in the entire codebase
rg "page" --type py -B 3 -A 3 | grep -A 5 -B 5 "aircan\|route\|add_url_rule"Repository: datopian/ckanext-aircan
Length of output: 948
🏁 Script executed:
# Check if there's any definition of the aircan.page route by searching more broadly
rg "page" ckanext/aircan/ --type pyRepository: datopian/ckanext-aircan
Length of output: 240
🏁 Script executed:
# Search for any Flask route decorators or additional blueprint definitions
rg "@.*\.route|def.*page" --type py ckanext/aircan/Repository: datopian/ckanext-aircan
Length of output: 49
🏁 Script executed:
# Check if there are any other files that might define routes
rg "Blueprint\|add_url_rule\|@.*route" --type py ckanext/aircan/Repository: datopian/ckanext-aircan
Length of output: 49
🏁 Script executed:
# Search for any route with "page" in the name or any simple route that might return "Hello, aircan!"
rg "Hello.*aircan" --type pyRepository: datopian/ckanext-aircan
Length of output: 233
🏁 Script executed:
# Also check if there might be another views file or if routes are defined elsewhere
find . -name "*.py" -type f | xargs grep -l "Hello.*aircan" 2>/dev/nullRepository: datopian/ckanext-aircan
Length of output: 142
🏁 Script executed:
# Let's search more broadly for blueprint routes or any mention of "page" as an endpoint name
rg "page.*view|endpoint.*page" --type py ckanext/aircan/Repository: datopian/ckanext-aircan
Length of output: 49
🏁 Script executed:
# Check the helpers.py file since it has aircan_hello
cat -n ckanext/aircan/helpers.pyRepository: datopian/ckanext-aircan
Length of output: 1838
🏁 Script executed:
# Let's also check the entire views.py again to see if we missed a "page" route
rg -n "add_url_rule\|@.*route" ckanext/aircan/views.pyRepository: datopian/ckanext-aircan
Length of output: 49
🏁 Script executed:
# Search for any test fixtures that might mock the response
cat ckanext/aircan/tests/conftest.py 2>/dev/null || echo "No conftest.py found"Repository: datopian/ckanext-aircan
Length of output: 87
The aircan.page route does not exist in the codebase.
The test references an endpoint that is never defined. The only route registered is /dataset/<id>/resource_pipeline/<resource_id> with endpoint name resource_pipeline (lines 97-99 of views.py). This route renders the resource_pipeline.html template with complex status data (lines 87-94), not the simple "Hello, aircan!" string the test expects.
Either implement the missing aircan.page route to return "Hello, aircan!" or update the test to verify the actual resource_pipeline route with proper parameters and assertions.
Additionally, the reset_db parameter is unused in the test body and should be removed.
🧰 Tools
🪛 Ruff (0.14.10)
13-13: Unused function argument: reset_db
(ARG001)
🤖 Prompt for AI Agents
In @ckanext/aircan/tests/test_views.py around lines 13 - 16, The test function
test_aircan_blueprint references a non-existent endpoint "aircan.page" and
expects "Hello, aircan!" while the codebase only registers the resource_pipeline
route (endpoint name "resource_pipeline") that renders resource_pipeline.html
with status data; update the test to call the existing endpoint (use
app.get(tk.h.url_for("resource_pipeline", id=<valid_dataset_id>,
resource_id=<valid_resource_id>)) or direct path
/dataset/<id>/resource_pipeline/<resource_id>), assert resp.status_code == 200
and check for expected content from resource_pipeline.html (or use
template/assertions that match the rendered status data), and remove the unused
reset_db parameter from the test signature; alternatively, if you prefer to keep
the original test behavior implement a new route/endpoint "aircan.page" in
views.py that returns "Hello, aircan!" and then keep the test but still remove
the unused reset_db param.
| classifiers = [ | ||
| "Development Status :: 4 - Beta", | ||
| "License :: OSI Approved :: GNU Affero General Public License v3 or later (AGPLv3+)", | ||
| "Programming Language :: Python :: 3.8", |
There was a problem hiding this comment.
🧩 Analysis chain
🌐 Web query:
Python 3.8 end of life EOL date
💡 Result:
Python 3.8 reached end-of-life on October 7, 2024. [1][2]
Sources:
[1] PEP 569 — Python 3.8 Release Schedule.
[2] Python EOL listings (eosl.date).
Remove Python 3.8 from supported versions. Python 3.8 reached EOL on October 7, 2024 and no longer receives security updates.
🤖 Prompt for AI Agents
In @pyproject.toml at line 13, Remove the Python 3.8 classifier entry from
pyproject.toml by deleting the "Programming Language :: Python :: 3.8" line
(i.e., update the classifiers list to exclude that exact string) so the project
no longer advertises support for EOL Python 3.8.
| [](https://github.com/Datopian/ckanext-aircan/actions) | ||
|
|
||
| # ckanext-aircan | ||
| A CKAN extension that integrates Airflow orchestrating with CKAN. This extension allows you to trigger, monitor, and display the status and logs of Aiflow data ingestion flows directly from the CKAN interface. |
There was a problem hiding this comment.
Fix typo: "Aiflow" → "Airflow".
📝 Proposed fix
-A CKAN extension that integrates Airflow orchestrating with CKAN. This extension allows you to trigger, monitor, and display the status and logs of Aiflow data ingestion flows directly from the CKAN interface.
+A CKAN extension that integrates Airflow orchestration with CKAN. This extension allows you to trigger, monitor, and display the status and logs of Airflow data ingestion flows directly from the CKAN interface.📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| A CKAN extension that integrates Airflow orchestrating with CKAN. This extension allows you to trigger, monitor, and display the status and logs of Aiflow data ingestion flows directly from the CKAN interface. | |
| A CKAN extension that integrates Airflow orchestration with CKAN. This extension allows you to trigger, monitor, and display the status and logs of Airflow data ingestion flows directly from the CKAN interface. |
🧰 Tools
🪛 LanguageTool
[grammar] ~4-~4: Ensure spelling is correct
Context: ...tor, and display the status and logs of Aiflow data ingestion flows directly from the ...
(QB_NEW_EN_ORTHOGRAPHY_ERROR_IDS_1)
🤖 Prompt for AI Agents
In @README.md at line 4, Replace the misspelled "Aiflow" with "Airflow" in the
README sentence describing the extension; update the phrase "display the status
and logs of Aiflow data ingestion flows" to "display the status and logs of
Airflow data ingestion flows" so the documentation uses the correct product
name.
| ## Features | ||
| - **Trigger Prefect Flows**: Automatically or manually submit CKAN resources for processing via Prefect. | ||
| - **Status & Logs**: View the current status and logs of Prefect flow runs associated with CKAN resources. |
There was a problem hiding this comment.
Inconsistent terminology: Features mention "Prefect" but extension is for Airflow.
The Features section references "Prefect Flows" and "Prefect flow runs," but the rest of the documentation consistently refers to Airflow. This creates confusion about what orchestration system the extension actually supports.
📝 Proposed fix
## Features
-- **Trigger Prefect Flows**: Automatically or manually submit CKAN resources for processing via Prefect.
-- **Status & Logs**: View the current status and logs of Prefect flow runs associated with CKAN resources.
+- **Trigger Airflow DAGs**: Automatically or manually submit CKAN resources for processing via Airflow.
+- **Status & Logs**: View the current status and logs of Airflow DAG runs associated with CKAN resources.📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| ## Features | |
| - **Trigger Prefect Flows**: Automatically or manually submit CKAN resources for processing via Prefect. | |
| - **Status & Logs**: View the current status and logs of Prefect flow runs associated with CKAN resources. | |
| ## Features | |
| - **Trigger Airflow DAGs**: Automatically or manually submit CKAN resources for processing via Airflow. | |
| - **Status & Logs**: View the current status and logs of Airflow DAG runs associated with CKAN resources. |
🤖 Prompt for AI Agents
In @README.md around lines 7 - 9, The Features section uses "Prefect" but the
project targets Airflow; update the two bullets under the "## Features" header
to replace "Prefect" with "Airflow" (e.g., change "Trigger Prefect Flows" to
"Trigger Airflow DAGs" and "Prefect flow runs" to "Airflow DAG runs"), and scan
the README for any other occurrences of "Prefect" and replace them with the
appropriate Airflow terminology to keep docs consistent.
There was a problem hiding this comment.
Actionable comments posted: 5
🤖 Fix all issues with AI agents
In @ckanext/aircan/logic/action.py:
- Around line 18-31: The function aircan_submit currently has a return type of
Dict[str, Any] but returns None when the resource format is not allowed; update
the function so it always returns a dict (e.g., return {} or return {"skipped":
True, "reason": "format not allowed"}), or alternatively change the signature to
Optional[Dict[str, Any]] if returning None is intended; locate the early-return
in the allowed_formats check and modify the return and type hint consistently
(references: function aircan_submit, allowed_formats).
- Around line 45-57: A new local DB session is created via
context["model"].meta.create_local_session() and stored into context with
context.update(...) but never closed; update the code that calls
create_local_session() (the context.update block) to either use a context
manager/with-style local session and ensure session.close() is called after
tk.get_action("aircan_status_update") finishes, or reuse the existing context
session instead of creating a new one; specifically modify the
create_local_session usage around the dag_run handling and the call to
aircan_status_update so the session is explicitly closed (or wrapped in a
with-statement) to avoid leaking DB connections.
- Around line 117-122: The code references p.toolkit.ValidationError which is
undefined; change that to use the existing toolkit alias (tk) by raising
tk.ValidationError instead of p.toolkit.ValidationError in the block that
validates resource_id (where data_dict.get("resource_id") and
tk.check_access("aircan_status_update", context, {"resource_id": resource_id})
are used) so the correct ValidationError symbol is referenced and no import is
required.
- Around line 62-64: In the except block catching requests.HTTPError, replace
the log.error call with log.exception(...) to capture the traceback, ensure you
return a Dict[str, Any] (e.g. {"success": False, "error": str(e)} or a localized
message) so the function matches its signature, and fix the undefined reference
to p.toolkit.ValidationError by using the correct toolkit symbol (replace
p.toolkit.ValidationError with tk.ValidationError or import the proper toolkit
alias) so the exception type is valid.
- Around line 33-39: The payload currently embeds the CKAN API key via
tk.config.get("ckanext.aircan.ckan_api_key") into the payload object (variable
payload) which is then sent to client.trigger_dag(); remove the raw key from
payload and instead include a reference (e.g., an Airflow Connection ID or
secret name) that the triggered DAG/task will use to retrieve the credential at
runtime from Airflow Connections or a secrets backend (Vault/AWS/GCP/K8s) or an
environment variable; update where client.trigger_dag() is called to pass only
the connection/secret identifier and modify the downstream task code to call
Airflow's get_connection or secrets backend APIs at runtime to fetch the actual
API key and use short-lived/least-privilege tokens.
🧹 Nitpick comments (3)
ckanext/aircan/logic/action.py (3)
84-100: Uselogging.exceptionfor better error diagnostics.When catching
requests.HTTPError, uselog.exception()instead oflog.error()to automatically include the full traceback. This helps with debugging Airflow connectivity issues.♻️ Proposed refactor
except requests.HTTPError as e: - log.error( + log.exception( tk._("Failed to fetch Airflow DAG run '%s': %s"), dag_run_id, str(e) )
134-151: Broad exception handling may hide important errors.The bare
except Exception:at line 147 catches all exceptions, which could mask unexpected errors (e.g., JSON decode errors, database issues). While the fallback is reasonable, consider being more specific or at least logging at WARNING level.♻️ Suggested improvement
- except Exception: + except (json.JSONDecodeError, KeyError, TypeError) as e: log.exception( "Failed to load previous aircan logs for resource_id=%s", resource_id )Or if you want to keep the broad catch, consider logging at WARNING level to make it more visible:
except Exception: - log.exception( + log.warning( "Failed to load previous aircan logs for resource_id=%s", resource_id )
1-179: Optional: Consider extracting magic strings as constants.Throughout the file, there are several magic strings like
"aircan","pipeline","queued","failed","info","error"that appear multiple times. Extracting these as module-level constants would improve maintainability and reduce typo risks.💡 Example refactor
At the top of the file, after imports:
# Task status constants TASK_TYPE_AIRCAN = "aircan" TASK_KEY_PIPELINE = "pipeline" STATE_QUEUED = "queued" STATE_FAILED = "failed" LOG_TYPE_INFO = "info" LOG_TYPE_ERROR = "error"Then use these throughout the code:
task_status = tk.get_action("task_status_show")( context, {"entity_id": resource_id, "task_type": TASK_TYPE_AIRCAN, "key": TASK_KEY_PIPELINE} )
📜 Review details
Configuration used: defaults
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (1)
ckanext/aircan/logic/action.py
🧰 Additional context used
🧬 Code graph analysis (1)
ckanext/aircan/logic/action.py (4)
ckanext/aircan/utils.py (1)
allowed_formats(17-28)ckanext/aircan/lib/airflow.py (3)
AirflowClient(16-153)trigger_dag(114-146)get_dag_run(148-153)ckanext/aircan/interfaces.py (2)
IAircan(4-14)update_payload(7-14)ckanext/aircan/plugin.py (2)
update_payload(79-86)get_actions(67-68)
🪛 Ruff (0.14.10)
ckanext/aircan/logic/action.py
63-63: Use logging.exception instead of logging.error
Replace with exception
(TRY400)
96-98: Use logging.exception instead of logging.error
Replace with exception
(TRY400)
120-120: Undefined name p
(F821)
🔇 Additional comments (1)
ckanext/aircan/logic/action.py (1)
174-179: LGTM!The
get_actions()function correctly follows the CKAN plugin pattern for exposing actions.
| payload = { | ||
| "resource": data_dict, | ||
| "ckan_config": { | ||
| "site_url": tk.config.get("ckan.site_url"), | ||
| "api_key": tk.config.get("ckanext.aircan.ckan_api_key"), | ||
| }, | ||
| } |
There was a problem hiding this comment.
🧩 Analysis chain
🏁 Script executed:
find . -name "action.py" -path "*/ckanext/aircan/*" | head -20Repository: datopian/ckanext-aircan
Length of output: 99
🏁 Script executed:
cat -n ckanext/aircan/logic/action.py | head -60Repository: datopian/ckanext-aircan
Length of output: 2371
🏁 Script executed:
# Search for where this payload is sent and how it's handled
rg -A 10 -B 5 "resource.*data_dict" ckanext/aircan/logic/action.pyRepository: datopian/ckanext-aircan
Length of output: 2648
🏁 Script executed:
# Look for any logging or debugging of this payload
rg "payload" ckanext/aircan/ -A 3 -B 3Repository: datopian/ckanext-aircan
Length of output: 4261
🌐 Web query:
CKAN API key security best practices external integrations
💡 Result:
Concise CKAN API-key/token security best practices for external integrations
- Use API tokens (not legacy API keys). Tokens are encrypted, revocable, and can expire; legacy keys are deprecated. [1][2]
- Issue one token per integration/service (don’t share user tokens). Name tokens clearly so they’re auditable. [1][2]
- Limit permissions: run integrations under separate service accounts/users with the minimum CKAN roles needed. Revoke UI/admin rights. [2]
- Enforce token expiration and automated rotation (use CKAN’s expire_api_token plugin or similar; rotate tokens regularly). [1][2]
- Transport and store tokens securely: always use TLS for API calls; store secrets in a secrets manager/vault or environment variables (never commit to code or client-side assets). [1][4]
- Use short-lived credentials or a token-exchange proxy when possible (avoid long-lived bearer tokens for public clients). [4]
- Restrict where tokens can be used: IP allowlists, network/firewall rules, and rate limits on endpoints where supported. [4]
- Audit, monitor and alert on token use: logging, usage dashboards, anomalous-usage alerts, and periodic reviews of active tokens. [4]
- Revoke compromised or unused tokens immediately and perform post‑incident review. [1][2]
- Prefer server-to-server integrations (backend-only) rather than embedding tokens in front-end or mobile apps; if front-end access is needed, use a secure intermediary service. [4]
Sources: CKAN docs & migration blog (API tokens, expire_api_token, deprecating legacy keys). [1][2]; general API-key security best practices (rotation, storage, monitoring). [4]
If you want, I can: 1) map these points to specific CKAN config/settings and commands on your CKAN version, or 2) draft a token-rotation policy and deploy checklist for your integrations. Which do you prefer?
🌐 Web query:
how to securely pass credentials to Airflow DAGs
💡 Result:
Short checklist + recommended options to securely provide credentials to Airflow DAGs:
-
Don’t hardcode credentials in DAG code or repo. Reference connection IDs or variable names instead. [1][7]
-
Use Airflow Connections (conn_id) from tasks/operators, not inline secrets. Populate Connections via:
- Secret backends (preferred) so Airflow reads secrets at runtime from a dedicated secrets manager. Supported backends: HashiCorp Vault, AWS Secrets Manager/SSM, GCP Secret Manager, Azure Key Vault, Kubernetes secrets, etc. Configure via AIRFLOW__SECRETS__BACKEND and backend kwargs. [7][4]
-
If you must keep secrets in Airflow storage:
- Enable and manage Fernet encryption (FERNET_KEY) so Connections and Variables are encrypted at rest; rotate keys using Airflow’s rotate-fernet-key procedure. Keep the Fernet key secure and out of version control. [2][6]
-
Use cloud/provider secret stores when possible:
- HashiCorp Vault backend (VaultBackend) — good for dynamic/short‑lived credentials and fine-grained auth. [9][11]
- AWS Secrets Manager / Systems Manager, GCP Secret Manager, Azure Key Vault — use provider backends so Airflow never stores plaintext in the metastore. Follow provider docs for service-account/role permissions. [7][4][5]
-
Minimize access and follow least privilege:
- Grant Airflow only the IAM/role permissions required to read the specific secrets; prefer instance/GKE/node/service accounts or workload identity over static credentials. [4][5]
-
Runtime considerations:
- Read secrets inside task execution (operator.execute) or templated fields, not at DAG-parse time; if webserver needs to render DAGs involving secrets, enable DAG serialization or ensure webserver has appropriate restricted access. [4]
-
Additional hygiene:
- Mask sensitive XCom/Log output and configure log masking.
- Don’t expose secret values in UI, logs, or worker environment dumps.
- Audit and rotate credentials periodically. [2][7]
Quick links for implementation details:
- Airflow Connections & AIRFLOW_CONN env vars. [1][3]
- Airflow secrets backends overview and provider list. [7]
- Fernet encryption & key rotation. [2][6]
- Cloud vendor guides (example: GCP Secret Manager / Cloud Composer). [4][5]
- HashiCorp Vault provider docs. [9]
If you tell me your platform (Kubernetes/Astro/Cloud Composer/AWS/GCP) I’ll give exact config snippets (AIRFLOW__SECRETS__BACKEND, backend kwargs, and examples).
API key exposed in Airflow DAG configuration — use Airflow secret backends instead.
The CKAN API key is being passed directly in the conf parameter to client.trigger_dag() (line 46), where it will be:
- Logged in Airflow's DAG run history and UI (visible to all Airflow users with dag read access)
- Stored unencrypted in Airflow's metastore
- Exposed if any dependent operator logs the configuration
- Visible in task logs and xcom outputs
Recommended solution: Use Airflow Connections and secret backends instead. Retrieve the API key at task runtime from:
- Airflow Connections (encrypted at rest with Fernet)
- A secret backend (HashiCorp Vault, AWS Secrets Manager, GCP Secret Manager, Kubernetes secrets, etc.)
- Environment variables on the worker/Airflow instance
This keeps the credential outside the DAG configuration and requires Airflow to read secrets from a dedicated secrets manager at runtime rather than embedding them in the payload. Pair this with short-lived tokens instead of the main API key, and limit the token's permissions to only what Airflow needs.
🤖 Prompt for AI Agents
In @ckanext/aircan/logic/action.py around lines 33 - 39, The payload currently
embeds the CKAN API key via tk.config.get("ckanext.aircan.ckan_api_key") into
the payload object (variable payload) which is then sent to
client.trigger_dag(); remove the raw key from payload and instead include a
reference (e.g., an Airflow Connection ID or secret name) that the triggered
DAG/task will use to retrieve the credential at runtime from Airflow Connections
or a secrets backend (Vault/AWS/GCP/K8s) or an environment variable; update
where client.trigger_dag() is called to pass only the connection/secret
identifier and modify the downstream task code to call Airflow's get_connection
or secrets backend APIs at runtime to fetch the actual API key and use
short-lived/least-privilege tokens.
| try: | ||
| dag_run = client.trigger_dag(conf=payload) | ||
| context.update({"session": context["model"].meta.create_local_session()}) | ||
| tk.get_action("aircan_status_update")( | ||
| context, | ||
| { | ||
| "resource_id": data_dict.get("id"), | ||
| "dag_run_id": dag_run.get("dag_run_id"), | ||
| "state": "queued", | ||
| "message": f"Added to the queue to be processed with '{dag_run.get("dag_run_id")}'.", | ||
| "clear_logs": True, | ||
| }, | ||
| ) |
There was a problem hiding this comment.
Potential resource leak: session not explicitly closed.
A new local session is created at line 47 but is never explicitly closed. While CKAN's session management may handle cleanup, it's better to use a context manager or ensure explicit cleanup to prevent potential resource leaks.
🔧 Suggested improvement
Consider whether the session creation is necessary here. If it is, document why a new local session is needed:
client = AirflowClient()
try:
dag_run = client.trigger_dag(conf=payload)
+ # Create local session for status update to avoid conflicting with parent transaction
context.update({"session": context["model"].meta.create_local_session()})
tk.get_action("aircan_status_update")(Alternatively, verify if the existing context session can be reused.
🤖 Prompt for AI Agents
In @ckanext/aircan/logic/action.py around lines 45 - 57, A new local DB session
is created via context["model"].meta.create_local_session() and stored into
context with context.update(...) but never closed; update the code that calls
create_local_session() (the context.update block) to either use a context
manager/with-style local session and ensure session.close() is called after
tk.get_action("aircan_status_update") finishes, or reuse the existing context
session instead of creating a new one; specifically modify the
create_local_session usage around the dag_run handling and the call to
aircan_status_update so the session is explicitly closed (or wrapped in a
with-statement) to avoid leaking DB connections.
There was a problem hiding this comment.
Actionable comments posted: 4
🤖 Fix all issues with AI agents
In @ckanext/aircan/logic/action.py:
- Around line 95-98: Replace the log.error call in the except block that catches
requests.HTTPError (the handler that logs "Failed to fetch Airflow DAG run '%s':
%s" and references dag_run_id) with log.exception so the traceback is included;
keep the existing formatted message using dag_run_id and the exception message
(or omit str(e) since log.exception will capture the exception), ensuring the
same context string is passed to log.exception in the except requests.HTTPError
as e: block.
- Around line 119-120: The ValidationError is raised using an undefined name
`p`; replace `p.toolkit.ValidationError` with the correct imported toolkit alias
`tk.ValidationError` so the check that raises when `resource_id` is missing uses
`tk.ValidationError` (same pattern as the rest of the file), e.g., update the
conditional that currently references `p` to call `tk.ValidationError` instead.
- Around line 62-63: In the except requests.HTTPError handler replace
log.error(...) with log.exception(...) to include the traceback (keep the
message context with client.dag_id), and then make the function return a defined
error value instead of implicit None—for example re-raise the exception or
return a consistent error dict like {'success': False, 'error': str(e)}; ensure
the chosen behavior matches the function's declared return type and callers'
expectations.
- Around line 25-31: The early exit in aircan_submit when
allowed_formats(resource_format) is false currently returns None; change it to
return a consistent dictionary matching the function's Dict[str, Any] contract
(e.g., include keys like "success": False, "reason": "format_not_allowed", "id":
data_dict.get("id"), and maybe "format": resource_format) so callers don't
receive None; update the branch that calls log.debug to return this informative
dict instead of implicit None and ensure any callers expecting a dict continue
to work.
🧹 Nitpick comments (2)
ckanext/aircan/logic/action.py (2)
65-66: Remove excessive blank lines.Multiple consecutive blank lines reduce code readability.
♻️ Proposed fix
- - - +
167-169: Document the coupling betweenclear_logsand error field clearing.Lines 167-169 unconditionally clear the error field when
clear_logs=True, regardless of whethertype="error". While this behavior appears intentional (resetting state when clearing logs), it is not documented in the function's docstring. Only one identified usage (inaircan_submitat line 55) passesclear_logs=True, and it does not usetype="error". Add clarification to the docstring to explain thatclear_logs=Trueclears both logs and error state, making this coupling explicit for future maintainers.
📜 Review details
Configuration used: defaults
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (2)
ckanext/aircan/logic/__init__.pyckanext/aircan/logic/action.py
🧰 Additional context used
🧬 Code graph analysis (1)
ckanext/aircan/logic/action.py (4)
ckanext/aircan/utils.py (1)
allowed_formats(17-28)ckanext/aircan/lib/airflow.py (3)
AirflowClient(16-153)trigger_dag(114-146)get_dag_run(148-153)ckanext/aircan/interfaces.py (2)
IAircan(4-14)update_payload(7-14)ckanext/aircan/plugin.py (2)
update_payload(79-86)get_actions(67-68)
🪛 Ruff (0.14.10)
ckanext/aircan/logic/action.py
63-63: Use logging.exception instead of logging.error
Replace with exception
(TRY400)
96-98: Use logging.exception instead of logging.error
Replace with exception
(TRY400)
120-120: Undefined name p
(F821)
🔇 Additional comments (2)
ckanext/aircan/logic/action.py (2)
174-179: LGTM!The action dispatcher correctly exposes all three actions.
47-47: Verify if local session creation after external API calls is necessary.The code explicitly creates a local session immediately after calling
client.trigger_dag()(external Airflow API) and before invokingaircan_status_update. This is the only occurrence ofcreate_local_sessionin the codebase and lacks documentation explaining the necessity.Confirm whether this session handling is required to isolate database operations from the previous session context following the external API call, or if the context already provides an appropriate session for
task_status_updateoperations.
| except requests.HTTPError as e: | ||
| log.error(tk._("Failed to trigger Airflow DAG '%s': %s"), client.dag_id, str(e)) |
There was a problem hiding this comment.
Use logging.exception for exception handlers and handle the error.
Two issues here:
- Use
logging.exceptioninstead oflogging.errorwithin exception handlers to automatically include the traceback. - After logging, the function returns
Noneimplicitly, which is inconsistent with the declared return type and could cause issues for callers.
Consider re-raising the exception or returning an error dictionary.
🔧 Proposed fix
except requests.HTTPError as e:
- log.error(tk._("Failed to trigger Airflow DAG '%s': %s"), client.dag_id, str(e))
+ log.exception(tk._("Failed to trigger Airflow DAG '%s': %s"), client.dag_id, str(e))
+ return {
+ "error": True,
+ "message": f"Failed to trigger Airflow DAG '{client.dag_id}': {str(e)}",
+ }📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| except requests.HTTPError as e: | |
| log.error(tk._("Failed to trigger Airflow DAG '%s': %s"), client.dag_id, str(e)) | |
| except requests.HTTPError as e: | |
| log.exception(tk._("Failed to trigger Airflow DAG '%s': %s"), client.dag_id, str(e)) | |
| return { | |
| "error": True, | |
| "message": f"Failed to trigger Airflow DAG '{client.dag_id}': {str(e)}", | |
| } |
🧰 Tools
🪛 Ruff (0.14.10)
63-63: Use logging.exception instead of logging.error
Replace with exception
(TRY400)
🤖 Prompt for AI Agents
In @ckanext/aircan/logic/action.py around lines 62 - 63, In the except
requests.HTTPError handler replace log.error(...) with log.exception(...) to
include the traceback (keep the message context with client.dag_id), and then
make the function return a defined error value instead of implicit None—for
example re-raise the exception or return a consistent error dict like
{'success': False, 'error': str(e)}; ensure the chosen behavior matches the
function's declared return type and callers' expectations.
| except requests.HTTPError as e: | ||
| log.error( | ||
| tk._("Failed to fetch Airflow DAG run '%s': %s"), dag_run_id, str(e) | ||
| ) |
There was a problem hiding this comment.
Use logging.exception in exception handler.
Within exception handlers, use logging.exception instead of logging.error to automatically include the traceback, which aids debugging.
🔧 Proposed fix
except requests.HTTPError as e:
- log.error(
+ log.exception(
tk._("Failed to fetch Airflow DAG run '%s': %s"), dag_run_id, str(e)
)📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| except requests.HTTPError as e: | |
| log.error( | |
| tk._("Failed to fetch Airflow DAG run '%s': %s"), dag_run_id, str(e) | |
| ) | |
| except requests.HTTPError as e: | |
| log.exception( | |
| tk._("Failed to fetch Airflow DAG run '%s': %s"), dag_run_id, str(e) | |
| ) |
🧰 Tools
🪛 Ruff (0.14.10)
96-98: Use logging.exception instead of logging.error
Replace with exception
(TRY400)
🤖 Prompt for AI Agents
In @ckanext/aircan/logic/action.py around lines 95 - 98, Replace the log.error
call in the except block that catches requests.HTTPError (the handler that logs
"Failed to fetch Airflow DAG run '%s': %s" and references dag_run_id) with
log.exception so the traceback is included; keep the existing formatted message
using dag_run_id and the exception message (or omit str(e) since log.exception
will capture the exception), ensuring the same context string is passed to
log.exception in the except requests.HTTPError as e: block.
There was a problem hiding this comment.
Actionable comments posted: 3
🤖 Fix all issues with AI agents
In @ckanext/aircan/logic/action.py:
- Around line 119-120: The code raises p.toolkit.ValidationError when
resource_id is missing but the symbol p is not imported; fix by either importing
ckan.plugins as p at the top of the module or by replacing
p.toolkit.ValidationError with toolkit.ValidationError after importing
ckan.plugins.toolkit as toolkit, then run tests to ensure the ValidationError is
correctly raised from the resource_id check in this function.
- Around line 25-31: The branch in aircan_submit that returns implicitly None
when not allowed_formats(resource_format) conflicts with the declared return
type Dict[str, Any]; either make the return type Optional[Dict[str, Any]] by
updating the function signature to include Optional and importing it, or return
a sentinel dict (e.g., {"skipped": True, "reason": "format not allowed", "id":
data_dict.get("id")}) so callers always receive a dict; update the
allowed_formats branch in the aircan_submit function accordingly and adjust any
callers if you choose the Optional approach.
- Around line 62-64: The except block that catches requests.HTTPError currently
logs with log.error and returns None, violating the Dict[str, Any] return type
and swallowing the exception; update the handler in the trigger_dag flow to call
log.exception(...) (so the traceback is preserved) and then either re-raise the
caught requests.HTTPError (raise) to propagate failure to callers or return a
well-typed error Dict (e.g., {'success': False, 'dag_id': client.dag_id,
'error': str(e)}) to honor the declared return type; ensure you reference the
existing client.dag_id and requests.HTTPError in the fix so the caller receives
an explicit failure signal instead of None.
🧹 Nitpick comments (2)
ckanext/aircan/logic/action.py (2)
1-16: Unused import:uuidThe
uuidmodule is imported but not used anywhere in this file. Consider removing it.Suggested fix
import logging -import uuid import json from datetime import datetime, timezone from typing import Any, Dict
95-98: Uselog.exceptionfor stack trace visibility.When catching the
HTTPError, usinglog.exceptioninstead oflog.errorwill include the stack trace, aiding debugging.Suggested fix
except requests.HTTPError as e: - log.error( + log.exception( tk._("Failed to fetch Airflow DAG run '%s': %s"), dag_run_id, str(e) )
📜 Review details
Configuration used: defaults
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (1)
ckanext/aircan/logic/action.py
🧰 Additional context used
🧬 Code graph analysis (1)
ckanext/aircan/logic/action.py (4)
ckanext/aircan/utils.py (1)
allowed_formats(17-28)ckanext/aircan/lib/airflow.py (3)
AirflowClient(16-153)trigger_dag(114-146)get_dag_run(148-153)ckanext/aircan/interfaces.py (2)
IAircan(4-14)update_payload(7-14)ckanext/aircan/plugin.py (2)
update_payload(79-86)get_actions(67-68)
🪛 Ruff (0.14.10)
ckanext/aircan/logic/action.py
63-63: Use logging.exception instead of logging.error
Replace with exception
(TRY400)
96-98: Use logging.exception instead of logging.error
Replace with exception
(TRY400)
120-120: Undefined name p
(F821)
🔇 Additional comments (1)
ckanext/aircan/logic/action.py (1)
174-179: LGTM!Clean dispatcher function that correctly exposes all three actions.
sagargg
force-pushed
the
feat/airflow-v3-support
branch
from
dce6ac3 to
d5b0ffb
Compare
There was a problem hiding this comment.
Actionable comments posted: 3
🤖 Fix all issues with AI agents
In @ckanext/aircan/logic/action.py:
- Around line 120-121: The code raises a ValidationError using an undefined name
`p` in the resource_id check; replace `p.toolkit.ValidationError` with the
correct symbol `tk.ValidationError` (the module imported as `tk`) so the line
raising the error uses `tk.ValidationError({"resource_id": ["Missing
resource_id"]})` for the `resource_id` validation; ensure any similar
occurrences use `tk` not `p`.
- Around line 63-65: The except block catching requests.HTTPError in the
function that calls trigger_dag should not silently return None; replace the
current log.error call with log.exception to record the stack trace and then
either re-raise a CKAN-friendly error (for example raise logic.ValidationError
or a descriptive Exception) or return a clear error dict (e.g., {"success":
False, "error": "Failed to trigger Airflow DAG: ..."}). Update the except clause
that references client.dag_id and str(e) to include the exception via
log.exception and then propagate the failure to callers via a raised
ValidationError or a structured error return so callers can distinguish
format/validation errors from Airflow HTTP failures.
- Line 31: The function in ckanext.aircan.logic.action that is annotated to
return Dict[str, Any] contains a bare "return" which yields None; replace that
bare return with a Dict-compatible value (e.g., return {} or return {"error":
"..."} as appropriate) or, if None is the intended signal, change the return
type to Optional[Dict[str, Any]] and update all callers to handle None; locate
the offending line with the bare "return" and apply one of these fixes
consistently with the function's callers and tests.
🧹 Nitpick comments (2)
ckanext/aircan/logic/action.py (2)
1-16: Unused import:uuidThe
uuidmodule is imported but never used in this file.Suggested fix
import logging -import uuid import json from datetime import datetime, timezone from typing import Any, Dict
96-99: Uselog.exceptionto preserve stack traceWhen catching the
HTTPError, usinglog.exceptioninstead oflog.errorwill automatically include the traceback, aiding debugging.Suggested fix
except requests.HTTPError as e: - log.error( + log.exception( tk._("Failed to fetch Airflow DAG run '%s': %s"), dag_run_id, str(e) )
📜 Review details
Configuration used: defaults
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (1)
ckanext/aircan/logic/action.py
🧰 Additional context used
🪛 Ruff (0.14.10)
ckanext/aircan/logic/action.py
59-62: Consider moving this statement to an else block
(TRY300)
64-64: Use logging.exception instead of logging.error
Replace with exception
(TRY400)
97-99: Use logging.exception instead of logging.error
Replace with exception
(TRY400)
121-121: Undefined name p
(F821)
🔇 Additional comments (1)
ckanext/aircan/logic/action.py (1)
175-180: LGTM!Standard CKAN action registration pattern.
| data_dict.get("id"), | ||
| resource_format, | ||
| ) | ||
| return |
There was a problem hiding this comment.
Return type violation: function returns None but type hint declares Dict[str, Any]
When the format is not allowed, the function implicitly returns None, which contradicts the declared return type. Callers expecting a dict may encounter AttributeError when accessing .get() on the result.
Suggested fix
)
- return
+ return {}Alternatively, update the type hint to Optional[Dict[str, Any]] and ensure all callers handle None.
📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| return | |
| return {} |
🤖 Prompt for AI Agents
In @ckanext/aircan/logic/action.py at line 31, The function in
ckanext.aircan.logic.action that is annotated to return Dict[str, Any] contains
a bare "return" which yields None; replace that bare return with a
Dict-compatible value (e.g., return {} or return {"error": "..."} as
appropriate) or, if None is the intended signal, change the return type to
Optional[Dict[str, Any]] and update all callers to handle None; locate the
offending line with the bare "return" and apply one of these fixes consistently
with the function's callers and tests.
| except requests.HTTPError as e: | ||
| log.error(tk._("Failed to trigger Airflow DAG '%s': %s"), client.dag_id, str(e)) | ||
|
|
There was a problem hiding this comment.
Silent failure on HTTPError — function returns None without signaling the error
When trigger_dag raises an HTTPError, the exception is logged but the function falls through and returns None. This causes two problems:
- Callers cannot distinguish "format not allowed" from "Airflow failure"
- Use
log.exceptionto preserve the stack trace
Suggested fix
except requests.HTTPError as e:
- log.error(tk._("Failed to trigger Airflow DAG '%s': %s"), client.dag_id, str(e))
+ log.exception(tk._("Failed to trigger Airflow DAG '%s': %s"), client.dag_id, str(e))
+ raiseOr return an error dict / raise a CKAN ValidationError with a user-friendly message if re-raising is undesirable.
📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| except requests.HTTPError as e: | |
| log.error(tk._("Failed to trigger Airflow DAG '%s': %s"), client.dag_id, str(e)) | |
| except requests.HTTPError as e: | |
| log.exception(tk._("Failed to trigger Airflow DAG '%s': %s"), client.dag_id, str(e)) | |
| raise |
🧰 Tools
🪛 Ruff (0.14.10)
64-64: Use logging.exception instead of logging.error
Replace with exception
(TRY400)
🤖 Prompt for AI Agents
In @ckanext/aircan/logic/action.py around lines 63 - 65, The except block
catching requests.HTTPError in the function that calls trigger_dag should not
silently return None; replace the current log.error call with log.exception to
record the stack trace and then either re-raise a CKAN-friendly error (for
example raise logic.ValidationError or a descriptive Exception) or return a
clear error dict (e.g., {"success": False, "error": "Failed to trigger Airflow
DAG: ..."}). Update the except clause that references client.dag_id and str(e)
to include the exception via log.exception and then propagate the failure to
callers via a raised ValidationError or a structured error return so callers can
distinguish format/validation errors from Airflow HTTP failures.
There was a problem hiding this comment.
Actionable comments posted: 2
🤖 Fix all issues with AI agents
In `@README.md`:
- Around line 18-20: Replace the TODO placeholder in README.md (the "**TODO:**
Add any additional install steps..." block) by either removing that entire
paragraph or replacing it with concrete installation steps or configuration
details required by the project (e.g., non-Python dependency install commands,
environment variables, or config file locations); ensure the README no longer
contains placeholder text and that any added steps are clear and complete for a
user following the install instructions.
- Line 14: The Requirements line currently reads "A running Airflow server or
Prefect Cloud" which mixes runtimes; update the README entry to reference only
Airflow (e.g., "A running Airflow server") or replace "Prefect Cloud" with the
correct Airflow-related requirement; edit the sentence containing that phrase in
README.md so it consistently refers to Airflow-only terminology.
♻️ Duplicate comments (7)
ckanext/aircan/logic/action.py (5)
113-116: Uselogging.exceptionin exception handler.Within exception handlers, use
log.exceptioninstead oflog.errorto automatically include the traceback for debugging.🔧 Proposed fix
except requests.HTTPError as e: - log.error( + log.exception( tk._("Failed to fetch Airflow DAG run '%s': %s"), dag_run_id, str(e) )
26-32: Return type violation: function returnsNonebut declaresDict[str, Any].When the format is not allowed, the function returns
Noneimplicitly viareturn, violating the declared return type. Callers expecting a dict may encounter errors.🐛 Proposed fix
data_dict.get("id"), resource_format, ) - return + return {}
34-39: API key embedded in payload sent to Airflow.The CKAN API key is passed directly in the DAG configuration payload, where it will be logged in Airflow's DAG run history/UI, stored in the metastore, and visible in task logs. Consider using Airflow Connections or secret backends to retrieve credentials at runtime instead.
64-84: Silent failure onHTTPError— function returnsNonewithout signaling error.When
trigger_dagraises anHTTPError, the exception is logged but execution falls through, returningNoneimplicitly. This violates the return type and prevents callers from distinguishing errors. Additionally, uselog.exceptionto preserve the stack trace.🐛 Proposed fix - re-raise as ValidationError
except requests.HTTPError as e: - log.error(tk._("Failed to trigger Airflow DAG '%s': %s"), client.dag_id, str(e)) + log.exception(tk._("Failed to trigger Airflow DAG '%s': %s"), client.dag_id, str(e)) + raise tk.ValidationError( + {"airflow": [f"Failed to trigger Airflow DAG: {str(e)}"]} + )
67-67: Session created but never explicitly closed.A new local session is created but not closed after use. Consider using a context manager or ensuring explicit cleanup to prevent resource leaks.
README.md (2)
4-4: Fix typo: "Aiflow" → "Airflow" and "orchestrating" → "orchestration".The description has a misspelled product name and grammatical error.
📝 Proposed fix
-A CKAN extension that integrates Airflow orchestrating with CKAN. This extension allows you to trigger, monitor, and display the status and logs of Aiflow data ingestion flows directly from the CKAN interface. +A CKAN extension that integrates Airflow orchestration with CKAN. This extension allows you to trigger, monitor, and display the status and logs of Airflow data ingestion flows directly from the CKAN interface.
7-9: Inconsistent terminology: "Prefect" should be "Airflow".The Features section references "Prefect Flows" but this extension is for Airflow.
📝 Proposed fix
## Features -- **Trigger Prefect Flows**: Automatically or manually submit CKAN resources for processing via Prefect. -- **Status & Logs**: View the current status and logs of Prefect flow runs associated with CKAN resources. +- **Trigger Airflow DAGs**: Automatically or manually submit CKAN resources for processing via Airflow. +- **Status & Logs**: View the current status and logs of Airflow DAG runs associated with CKAN resources.
🧹 Nitpick comments (2)
README.md (1)
44-67: Add language specifiers to fenced code blocks.Fenced code blocks should have a language specified for proper syntax highlighting and linting compliance.
📝 Proposed fix for config block (lines 44-67)
-``` +```ini ckanext.aircan.endpoint = http://localhost:8080📝 Proposed fix for plugin example (lines 119-128)
-``` +```python class ExamplePlugin(plugins.SingletonPlugin):Also applies to: 119-128
ckanext/aircan/logic/action.py (1)
165-168: Broad exception handling may mask specific errors.Catching bare
Exceptioncan hide unexpected errors. Consider catching more specific exceptions (e.g.,json.JSONDecodeError,tk.ObjectNotFound) or at minimum re-log with more context.🔧 Suggested improvement
- except Exception: + except (json.JSONDecodeError, tk.ObjectNotFound) as exc: log.exception( "Failed to load previous aircan logs for resource_id=%s", resource_id )Alternatively, if catching all exceptions is intentional, this is acceptable since
log.exceptionalready captures full context.
📜 Review details
Configuration used: defaults
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (2)
README.mdckanext/aircan/logic/action.py
🧰 Additional context used
🧬 Code graph analysis (1)
ckanext/aircan/logic/action.py (5)
ckanext/aircan/utils.py (1)
allowed_formats(17-28)ckanext/aircan/lib/airflow.py (3)
AirflowClient(16-153)trigger_dag(114-146)get_dag_run(148-153)ckanext/aircan/logic/auth.py (3)
aircan_submit(5-8)aircan_status(11-14)aircan_status_update(17-20)ckanext/aircan/interfaces.py (2)
IAircan(4-14)update_payload(7-14)ckanext/aircan/plugin.py (2)
update_payload(79-86)get_actions(67-68)
🪛 LanguageTool
README.md
[grammar] ~4-~4: Ensure spelling is correct
Context: ...tor, and display the status and logs of Aiflow data ingestion flows directly from the ...
(QB_NEW_EN_ORTHOGRAPHY_ERROR_IDS_1)
🪛 markdownlint-cli2 (0.18.1)
README.md
30-30: Bare URL used
(MD034, no-bare-urls)
44-44: Fenced code blocks should have a language specified
(MD040, fenced-code-language)
73-73: Code block style
Expected: fenced; Actual: indented
(MD046, code-block-style)
119-119: Fenced code blocks should have a language specified
(MD040, fenced-code-language)
🪛 Ruff (0.14.11)
ckanext/aircan/logic/action.py
78-81: Consider moving this statement to an else block
(TRY300)
83-83: Use logging.exception instead of logging.error
Replace with exception
(TRY400)
114-116: Use logging.exception instead of logging.error
Replace with exception
(TRY400)
🔇 Additional comments (2)
ckanext/aircan/logic/action.py (2)
135-138: LGTM:tk.ValidationErroris now correctly used.The previous issue with undefined
phas been fixed; the code now properly usestk.ValidationError.
192-197: LGTM!The
get_actionsfunction correctly exposes all three actions for CKAN's action interface.
✏️ Tip: You can disable this entire section by setting review_details to false in your review settings.
| ## Requirements | ||
| - CKAN 2.11 or later (not tested on earlier versions) | ||
| - Python 3.8+ | ||
| - A running [Airflow](https://airflow.apache.org/) server or Prefect Cloud |
There was a problem hiding this comment.
Requirements section also references "Prefect Cloud".
This is another instance of the Prefect/Airflow terminology inconsistency. The extension is for Airflow, so "Prefect Cloud" should be removed or replaced.
📝 Proposed fix
## Requirements
- CKAN 2.11 or later (not tested on earlier versions)
- Python 3.8+
-- A running [Airflow](https://airflow.apache.org/) server or Prefect Cloud
+- A running [Airflow](https://airflow.apache.org/) server🤖 Prompt for AI Agents
In `@README.md` at line 14, The Requirements line currently reads "A running
Airflow server or Prefect Cloud" which mixes runtimes; update the README entry
to reference only Airflow (e.g., "A running Airflow server") or replace "Prefect
Cloud" with the correct Airflow-related requirement; edit the sentence
containing that phrase in README.md so it consistently refers to Airflow-only
terminology.
| **TODO:** Add any additional install steps to the list below. | ||
| For example installing any non-Python dependencies or adding any required | ||
| config settings. |
There was a problem hiding this comment.
Remove or complete the TODO placeholder.
Installation instructions should not contain placeholder text for users. Either complete the additional install steps or remove this block.
📝 Proposed fix
-**TODO:** Add any additional install steps to the list below.
-For example installing any non-Python dependencies or adding any required
-config settings.
-
To install ckanext-aircan:📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| **TODO:** Add any additional install steps to the list below. | |
| For example installing any non-Python dependencies or adding any required | |
| config settings. |
🤖 Prompt for AI Agents
In `@README.md` around lines 18 - 20, Replace the TODO placeholder in README.md
(the "**TODO:** Add any additional install steps..." block) by either removing
that entire paragraph or replacing it with concrete installation steps or
configuration details required by the project (e.g., non-Python dependency
install commands, environment variables, or config file locations); ensure the
README no longer contains placeholder text and that any added steps are clear
and complete for a user following the install instructions.
Summary by CodeRabbit
✏️ Tip: You can customize this high-level summary in your review settings.