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

fix return values on glue operators deferrable mode #31694

Merged
merged 2 commits into from
Jun 5, 2023
Merged

fix return values on glue operators deferrable mode #31694

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
08bdfaa
fix return values on glue operators deferrable mode
vandonr-amz Jun 2, 2023
703cfe5
don't fetch value from dict when it's right there
vandonr-amz Jun 5, 2023
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
2 changes: 1 addition & 1 deletion airflow/providers/amazon/aws/operators/glue.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -200,4 +200,4 @@ def execute(self, context: Context):
def execute_complete(self, context, event=None):
if event["status"] != "success":
raise AirflowException(f"Error in glue job: {event}")
return
return event["value"]
2 changes: 1 addition & 1 deletion airflow/providers/amazon/aws/operators/glue_crawler.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -107,4 +107,4 @@ def execute(self, context: Context):
def execute_complete(self, context, event=None):
if event["status"] != "success":
raise AirflowException(f"Error in glue crawl: {event}")
return
return self.config["Name"]
4 changes: 2 additions & 2 deletions airflow/providers/amazon/aws/triggers/glue.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -59,5 +59,5 @@ def serialize(self) -> tuple[str, dict[str, Any]]:

async def run(self) -> AsyncIterator[TriggerEvent]:
hook = GlueJobHook(aws_conn_id=self.aws_conn_id)
await hook.async_job_completion(self.job_name, self.run_id, self.verbose)
yield TriggerEvent({"status": "success", "message": "Job done"})
glue_job_run = await hook.async_job_completion(self.job_name, self.run_id, self.verbose)
yield TriggerEvent({"status": "success", "message": "Job done", "value": glue_job_run["JobRunId"]})
Copy link
Member

@pankajkoti pankajkoti Jun 4, 2023 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Isn't glue_job_run["JobRunId"] the same as self.run_id?

Perhaps we could just return that, so no need to fetch the value from the dictionary key.
Upto if you'd like to consider the suggestion :)

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

hmm yes you're right :p

5 changes: 1 addition & 4 deletions airflow/providers/amazon/aws/triggers/glue_crawler.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -68,10 +68,7 @@ async def run(self) -> AsyncIterator[TriggerEvent]:
break # we reach this point only if the waiter met a success criteria
except WaiterError as error:
if "terminal failure" in str(error):
yield TriggerEvent(
{"status": "failure", "message": f"Glue Crawler creation Failed: {error}"}
)
break
raise
self.log.info("Status of glue crawl is %s", error.last_response["Crawler"]["State"])
await asyncio.sleep(int(self.poll_interval))

Expand Down