Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

ISSUE-19254: Fix pipeline unable to run in current day/week/month error #19523

Open
wants to merge 3 commits into
base: main
Choose a base branch
from
Open

ISSUE-19254: Fix pipeline unable to run in current day/week/month error #19523

Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
127ac9d
ISSUE-19254: Fix pipeline unable to run in current day/week/month error
datapopcorn Jan 24, 2025
42b2728
ISSUE-19254: Fix on demand job error
datapopcorn Jan 24, 2025
9c5f521
fix py_format
datapopcorn Jan 24, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
19 changes: 14 additions & 5 deletions openmetadata-airflow-apis/openmetadata_managed_apis/workflows/ingestion/common.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -44,6 +44,7 @@
except ModuleNotFoundError:
from airflow.operators.python_operator import PythonOperator

from croniter import croniter
from openmetadata_managed_apis.utils.logger import set_operator_logger, workflow_logger
from openmetadata_managed_apis.utils.parser import (
parse_service_connection,
Expand Down Expand Up @@ -247,11 +248,19 @@ def build_dag_configs(ingestion_pipeline: IngestionPipeline) -> dict:
:param ingestion_pipeline: pipeline configs
:return: dict to use as kwargs
"""

if ingestion_pipeline.airflowConfig.startDate:
start_date = ingestion_pipeline.airflowConfig.startDate.root
# Determine start_date based on schedule_interval using croniter
schedule_interval = ingestion_pipeline.airflowConfig.scheduleInterval
now = datetime.now()

if schedule_interval is None:
# On-demand DAG, set start_date to now
start_date = now
elif croniter.is_valid(schedule_interval):
cron = croniter(schedule_interval, now)
start_date = cron.get_prev(datetime)
else:
start_date = datetime.now() - timedelta(days=1)
# Handle invalid cron expressions if necessary
start_date = now

return {
"dag_id": clean_dag_id(ingestion_pipeline.name.root),
Expand All @@ -272,7 +281,7 @@ def build_dag_configs(ingestion_pipeline: IngestionPipeline) -> dict:
"is_paused_upon_creation": ingestion_pipeline.airflowConfig.pausePipeline
or False,
"catchup": ingestion_pipeline.airflowConfig.pipelineCatchup or False,
"schedule_interval": ingestion_pipeline.airflowConfig.scheduleInterval,
"schedule_interval": schedule_interval,
"tags": [
"OpenMetadata",
clean_name_tag(ingestion_pipeline.displayName)
Expand Down
Loading