You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by GitBox <gi...@apache.org> on 2020/06/23 00:58:36 UTC
[GitHub] [airflow] ephraimbuddy opened a new pull request #9482: add crud endpoint for xcom
ephraimbuddy opened a new pull request #9482:
URL: https://github.com/apache/airflow/pull/9482
---
Closes : #9111
Make sure to mark the boxes below before creating PR: [x]
- [ ] Description above provides context of the change
- [ ] Unit tests coverage for changes (not needed for documentation changes)
- [ ] Target Github ISSUE in description if exists
- [ ] Commits follow "[How to write a good git commit message](http://chris.beams.io/posts/git-commit/)"
- [ ] Relevant documentation is updated including usage instructions.
- [ ] I will engage committers as explained in [Contribution Workflow Example](https://github.com/apache/airflow/blob/master/CONTRIBUTING.rst#contribution-workflow-example).
---
In case of fundamental code change, Airflow Improvement Proposal ([AIP](https://cwiki.apache.org/confluence/display/AIRFLOW/Airflow+Improvements+Proposals)) is needed.
In case of a new dependency, check compliance with the [ASF 3rd Party License Policy](https://www.apache.org/legal/resolved.html#category-x).
In case of backwards incompatible changes please leave a note in [UPDATING.md](https://github.com/apache/airflow/blob/master/UPDATING.md).
Read the [Pull Request Guidelines](https://github.com/apache/airflow/blob/master/CONTRIBUTING.rst#pull-request-guidelines) for more information.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] ephraimbuddy commented on pull request #9482: Add CRUD endpoint for XCom
Posted by GitBox <gi...@apache.org>.
ephraimbuddy commented on pull request #9482:
URL: https://github.com/apache/airflow/pull/9482#issuecomment-653119162
Hi, @turbaszek can I ask for review?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] ephraimbuddy closed pull request #9482: Add CRUD endpoint for XCom
Posted by GitBox <gi...@apache.org>.
ephraimbuddy closed pull request #9482:
URL: https://github.com/apache/airflow/pull/9482
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] turbaszek commented on a change in pull request #9482: Add CRUD endpoint for XCom
Posted by GitBox <gi...@apache.org>.
turbaszek commented on a change in pull request #9482:
URL: https://github.com/apache/airflow/pull/9482#discussion_r449452345
##########
File path: airflow/api_connexion/endpoints/xcom_endpoint.py
##########
@@ -89,15 +111,78 @@ def get_xcom_entry(
return xcom_collection_item_schema.dump(query_object)
-def patch_xcom_entry():
+@provide_session
+def patch_xcom_entry(
+ session: Session,
+ dag_id: str,
+ dag_run_id: str,
+ task_id: str,
+ xcom_key: str,
+ update_mask: Optional[List[str]] = None
+):
"""
Update an XCom entry
"""
- raise NotImplementedError("Not implemented yet.")
+ dag_run = session.query(DR).filter(DR.run_id == dag_run_id,
+ DR.dag_id == dag_id).first()
+ if not dag_run:
+ raise NotFound(f'DAGRun with dag_id:{dag_id} and run_id:{dag_run_id} not found')
+
+ xcom_entry = session.query(XCom).filter(XCom.dag_id == dag_id,
+ XCom.task_id == task_id,
+ XCom.key == xcom_key).first()
+
+ if not xcom_entry:
+ raise NotFound("XCom not found")
+ try:
+ body = xcom_schema.load(request.json, partial=("dag_id", "task_id",
+ "key", "execution_date"))
+ except ValidationError as err:
+ raise BadRequest(detail=err.messages.get('_schema', [str(err.messages)])[0])
+ data = body.data
+ # Check that other attributes are not being updated
+ for field in data:
+ if field != 'value' and data[field] != getattr(xcom_entry, field):
Review comment:
I think this may result in error if there's no `field` attribute of `xcom_entry`.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] turbaszek commented on a change in pull request #9482: Add CRUD endpoint for XCom
Posted by GitBox <gi...@apache.org>.
turbaszek commented on a change in pull request #9482:
URL: https://github.com/apache/airflow/pull/9482#discussion_r449455360
##########
File path: airflow/api_connexion/endpoints/xcom_endpoint.py
##########
@@ -89,15 +111,78 @@ def get_xcom_entry(
return xcom_collection_item_schema.dump(query_object)
-def patch_xcom_entry():
+@provide_session
+def patch_xcom_entry(
+ session: Session,
+ dag_id: str,
+ dag_run_id: str,
+ task_id: str,
+ xcom_key: str,
+ update_mask: Optional[List[str]] = None
+):
"""
Update an XCom entry
"""
- raise NotImplementedError("Not implemented yet.")
+ dag_run = session.query(DR).filter(DR.run_id == dag_run_id,
+ DR.dag_id == dag_id).first()
+ if not dag_run:
+ raise NotFound(f'DAGRun with dag_id:{dag_id} and run_id:{dag_run_id} not found')
+
+ xcom_entry = session.query(XCom).filter(XCom.dag_id == dag_id,
+ XCom.task_id == task_id,
+ XCom.key == xcom_key).first()
+
+ if not xcom_entry:
+ raise NotFound("XCom not found")
+ try:
+ body = xcom_schema.load(request.json, partial=("dag_id", "task_id",
+ "key", "execution_date"))
+ except ValidationError as err:
+ raise BadRequest(detail=err.messages.get('_schema', [str(err.messages)])[0])
+ data = body.data
+ # Check that other attributes are not being updated
+ for field in data:
+ if field != 'value' and data[field] != getattr(xcom_entry, field):
+ raise BadRequest(detail=f"'{field}' cannot be updated")
+ if update_mask:
+ update_mask = [i.strip() for i in update_mask]
+ data_ = {}
+ for field in update_mask:
+ if field in data and field != 'value':
+ raise BadRequest(detail=f"{field} cannot be updated")
+ elif field in data and field == 'value':
+ data_[field] = data[field]
+ else:
+ raise BadRequest(detail=f"Unknown field '{field}' in update_mask")
+ data = data_
+ data['value'] = XCom.serialize_value(data['value'])
+ xcom_entry.value = data['value']
+ session.add(xcom_entry)
+ session.commit()
+ return xcom_schema.dump(xcom_entry)
Review comment:
Commit is performed by `@provide_session` and I'm wondering if we need another one here explicit. I think using the one from `provide_session` has the advantage that if returning response will fail then there will be no change on db level. WDYT? @mik-laj @feluelle @potiuk
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] feluelle commented on a change in pull request #9482: Add CRUD endpoint for XCom
Posted by GitBox <gi...@apache.org>.
feluelle commented on a change in pull request #9482:
URL: https://github.com/apache/airflow/pull/9482#discussion_r449463417
##########
File path: airflow/api_connexion/endpoints/xcom_endpoint.py
##########
@@ -89,15 +111,78 @@ def get_xcom_entry(
return xcom_collection_item_schema.dump(query_object)
-def patch_xcom_entry():
+@provide_session
+def patch_xcom_entry(
+ session: Session,
+ dag_id: str,
+ dag_run_id: str,
+ task_id: str,
+ xcom_key: str,
+ update_mask: Optional[List[str]] = None
+):
"""
Update an XCom entry
"""
- raise NotImplementedError("Not implemented yet.")
+ dag_run = session.query(DR).filter(DR.run_id == dag_run_id,
+ DR.dag_id == dag_id).first()
+ if not dag_run:
+ raise NotFound(f'DAGRun with dag_id:{dag_id} and run_id:{dag_run_id} not found')
+
+ xcom_entry = session.query(XCom).filter(XCom.dag_id == dag_id,
+ XCom.task_id == task_id,
+ XCom.key == xcom_key).first()
+
+ if not xcom_entry:
+ raise NotFound("XCom not found")
+ try:
+ body = xcom_schema.load(request.json, partial=("dag_id", "task_id",
+ "key", "execution_date"))
+ except ValidationError as err:
+ raise BadRequest(detail=err.messages.get('_schema', [str(err.messages)])[0])
+ data = body.data
+ # Check that other attributes are not being updated
+ for field in data:
+ if field != 'value' and data[field] != getattr(xcom_entry, field):
+ raise BadRequest(detail=f"'{field}' cannot be updated")
+ if update_mask:
+ update_mask = [i.strip() for i in update_mask]
+ data_ = {}
+ for field in update_mask:
+ if field in data and field != 'value':
+ raise BadRequest(detail=f"{field} cannot be updated")
+ elif field in data and field == 'value':
+ data_[field] = data[field]
+ else:
+ raise BadRequest(detail=f"Unknown field '{field}' in update_mask")
+ data = data_
+ data['value'] = XCom.serialize_value(data['value'])
+ xcom_entry.value = data['value']
+ session.add(xcom_entry)
+ session.commit()
+ return xcom_schema.dump(xcom_entry)
Review comment:
> [...] then there will be no change on db level.
Which is correct. So I am for removing the manual commit from there.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] ephraimbuddy commented on a change in pull request #9482: Add CRUD endpoint for XCom
Posted by GitBox <gi...@apache.org>.
ephraimbuddy commented on a change in pull request #9482:
URL: https://github.com/apache/airflow/pull/9482#discussion_r445524042
##########
File path: airflow/api_connexion/endpoints/xcom_endpoint.py
##########
@@ -14,25 +14,47 @@
# KIND, either express or implied. See the License for the
# specific language governing permissions and limitations
# under the License.
-from flask import request
+from typing import List, Optional
+
+from flask import Response, request
+from marshmallow import ValidationError
from sqlalchemy import and_, func
from sqlalchemy.orm.session import Session
from airflow.api_connexion import parameters
-from airflow.api_connexion.exceptions import NotFound
+from airflow.api_connexion.exceptions import BadRequest, NotFound
from airflow.api_connexion.schemas.xcom_schema import (
XComCollection, XComCollectionItemSchema, XComCollectionSchema, xcom_collection_item_schema,
- xcom_collection_schema,
+ xcom_collection_schema, xcom_schema,
)
from airflow.models import DagRun as DR, XCom
from airflow.utils.session import provide_session
-def delete_xcom_entry():
+@provide_session
+def delete_xcom_entry(
+ dag_id: str,
+ dag_run_id: str,
+ task_id: str,
+ xcom_key: str,
+ session: Session
+) -> Response:
"""
Delete an XCom entry
"""
- raise NotImplementedError("Not implemented yet.")
+
+ dag_run = session.query(DR).filter(DR.run_id == dag_run_id,
+ DR.dag_id == dag_id).first()
+ if not dag_run:
+ raise NotFound(f'DAGRun with dag_id:{dag_id} and run_id:{dag_run_id} not found')
+
+ entry = session.query(XCom).filter(XCom.dag_id == dag_id,
+ XCom.task_id == task_id,
+ XCom.key == xcom_key).delete()
+ if not entry:
Review comment:
Yeah. The suggestion is good especially when the query is not long. Here, the query is long and according to the Zen of python, Readability counts, that's why I assigned the number of rows deleted to entry and check it in the next line.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] mik-laj commented on pull request #9482: Add CRUD endpoint for XCom
Posted by GitBox <gi...@apache.org>.
mik-laj commented on pull request #9482:
URL: https://github.com/apache/airflow/pull/9482#issuecomment-653434020
@turbaszek @feluelle @ephraimbuddy
I have doubts whether we should allow third-party systems to create and update, delete entries. I don't think this is expected because we store metadata from the execution of a specific task there. We don't want other systems to be able to affect task execution. For customizable configuration, we have variables. The user will be able to delete XCOM entries by cleaning the task instance. WDYT?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] turbaszek commented on a change in pull request #9482: Add CRUD endpoint for XCom
Posted by GitBox <gi...@apache.org>.
turbaszek commented on a change in pull request #9482:
URL: https://github.com/apache/airflow/pull/9482#discussion_r449454085
##########
File path: airflow/api_connexion/endpoints/xcom_endpoint.py
##########
@@ -89,15 +111,78 @@ def get_xcom_entry(
return xcom_collection_item_schema.dump(query_object)
-def patch_xcom_entry():
+@provide_session
+def patch_xcom_entry(
+ session: Session,
+ dag_id: str,
+ dag_run_id: str,
+ task_id: str,
+ xcom_key: str,
+ update_mask: Optional[List[str]] = None
+):
"""
Update an XCom entry
"""
- raise NotImplementedError("Not implemented yet.")
+ dag_run = session.query(DR).filter(DR.run_id == dag_run_id,
+ DR.dag_id == dag_id).first()
+ if not dag_run:
+ raise NotFound(f'DAGRun with dag_id:{dag_id} and run_id:{dag_run_id} not found')
+
+ xcom_entry = session.query(XCom).filter(XCom.dag_id == dag_id,
+ XCom.task_id == task_id,
+ XCom.key == xcom_key).first()
+
+ if not xcom_entry:
+ raise NotFound("XCom not found")
+ try:
+ body = xcom_schema.load(request.json, partial=("dag_id", "task_id",
+ "key", "execution_date"))
+ except ValidationError as err:
+ raise BadRequest(detail=err.messages.get('_schema', [str(err.messages)])[0])
+ data = body.data
+ # Check that other attributes are not being updated
+ for field in data:
+ if field != 'value' and data[field] != getattr(xcom_entry, field):
+ raise BadRequest(detail=f"'{field}' cannot be updated")
+ if update_mask:
+ update_mask = [i.strip() for i in update_mask]
+ data_ = {}
+ for field in update_mask:
+ if field in data and field != 'value':
+ raise BadRequest(detail=f"{field} cannot be updated")
+ elif field in data and field == 'value':
+ data_[field] = data[field]
+ else:
+ raise BadRequest(detail=f"Unknown field '{field}' in update_mask")
Review comment:
```suggestion
if field not in data:
raise BadRequest(detail=f"Unknown field '{field}' in update_mask")
elif field != 'value':
raise BadRequest(detail=f"{field} cannot be updated")
elif field == 'value':
data_[field] = data[field]
```
WDYT?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] mik-laj edited a comment on pull request #9482: Add CRUD endpoint for XCom
Posted by GitBox <gi...@apache.org>.
mik-laj edited a comment on pull request #9482:
URL: https://github.com/apache/airflow/pull/9482#issuecomment-653440785
@feluelle These API will not address this problem. I want to build an API for third systems, not an internal API. In my opinion, API for third-party systems should maintain backward compatibility and should provide access to the most important operations. Building an API that is too rich in features or provides too much information causes problems with maintaining backward compatibility.
I would not want this API to be only access to the database via HTTP, but it was an API that will easily allow you to build new applications on it, e.g. applications for monitoring multiple instances of Airflow and provide mult-tenant view.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] kaxil commented on a change in pull request #9482: add crud endpoint for xcom
Posted by GitBox <gi...@apache.org>.
kaxil commented on a change in pull request #9482:
URL: https://github.com/apache/airflow/pull/9482#discussion_r443906611
##########
File path: airflow/api_connexion/endpoints/xcom_endpoint.py
##########
@@ -14,16 +14,45 @@
# KIND, either express or implied. See the License for the
# specific language governing permissions and limitations
# under the License.
+from typing import List, Optional
-# TODO(mik-laj): We have to implement it.
-# Do you want to help? Please look at: sshttps://github.com/apache/airflow/issues/8134
+from flask import Response, request
+from marshmallow import ValidationError
+from sqlalchemy.orm.session import Session
+from airflow.api_connexion.exceptions import AlreadyExists, BadRequest, NotFound
+from airflow.api_connexion.schemas.xcom_schema import xcom_schema
+from airflow.models import DagRun as DR, XCom
+from airflow.utils.session import provide_session
-def delete_xcom_entry():
+
+@provide_session
+def delete_xcom_entry(
+ dag_id: str,
+ dag_run_id: str,
+ task_id: str,
+ xcom_key: str,
+ session: Session
+) -> Response:
"""
Delete an XCom entry
"""
- raise NotImplementedError("Not implemented yet.")
+
+ dag_run = session.query(DR).filter(DR.run_id == dag_run_id,
+ DR.dag_id == dag_id).first()
+ if not dag_run:
+ raise NotFound(f'DAGRun with dag_id:{dag_id} and run_id:{dag_run_id} not found')
+
+ query = session.query(XCom)
+ query = query.filter(XCom.dag_id == dag_id,
+ XCom.task_id == task_id,
+ XCom.key == xcom_key)
+ entry = query.delete()
Review comment:
You can chain these SQL Queries as I have done here: https://github.com/apache/airflow/pull/9424
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] turbaszek commented on pull request #9482: Add CRUD endpoint for XCom
Posted by GitBox <gi...@apache.org>.
turbaszek commented on pull request #9482:
URL: https://github.com/apache/airflow/pull/9482#issuecomment-653465013
> I have doubts whether we should allow third-party systems to create and update, delete entries. I don't think this is expected because **we store metadata** from the execution of a specific task there. We don't want other systems to be able to **affect task execution**. For customizable configuration, we have variables.
I agree here. I think that users should not be able to update XCom, TaskInstance and other "task execution-related" entities.
> Hm.. I could imagine having multiple airflow instances which might want to share those xcom values with each other. WDYT?
I think in this case read only access should be enough.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] OmairK commented on a change in pull request #9482: Add CRUD endpoint for XCom
Posted by GitBox <gi...@apache.org>.
OmairK commented on a change in pull request #9482:
URL: https://github.com/apache/airflow/pull/9482#discussion_r445513453
##########
File path: airflow/api_connexion/endpoints/xcom_endpoint.py
##########
@@ -14,25 +14,47 @@
# KIND, either express or implied. See the License for the
# specific language governing permissions and limitations
# under the License.
-from flask import request
+from typing import List, Optional
+
+from flask import Response, request
+from marshmallow import ValidationError
from sqlalchemy import and_, func
from sqlalchemy.orm.session import Session
from airflow.api_connexion import parameters
-from airflow.api_connexion.exceptions import NotFound
+from airflow.api_connexion.exceptions import BadRequest, NotFound
from airflow.api_connexion.schemas.xcom_schema import (
XComCollection, XComCollectionItemSchema, XComCollectionSchema, xcom_collection_item_schema,
- xcom_collection_schema,
+ xcom_collection_schema, xcom_schema,
)
from airflow.models import DagRun as DR, XCom
from airflow.utils.session import provide_session
-def delete_xcom_entry():
+@provide_session
+def delete_xcom_entry(
+ dag_id: str,
+ dag_run_id: str,
+ task_id: str,
+ xcom_key: str,
+ session: Session
+) -> Response:
"""
Delete an XCom entry
"""
- raise NotImplementedError("Not implemented yet.")
+
+ dag_run = session.query(DR).filter(DR.run_id == dag_run_id,
+ DR.dag_id == dag_id).first()
+ if not dag_run:
+ raise NotFound(f'DAGRun with dag_id:{dag_id} and run_id:{dag_run_id} not found')
+
+ entry = session.query(XCom).filter(XCom.dag_id == dag_id,
+ XCom.task_id == task_id,
+ XCom.key == xcom_key).delete()
+ if not entry:
Review comment:
```suggestion
if session.query(XCom).filter(XCom.dag_id == dag_id,
XCom.task_id == task_id,
XCom.key == xcom_key).delete() == 0:
```
You could do something like this to make the code shorter, delete returns the number of rows deleted
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] mik-laj commented on pull request #9482: Add CRUD endpoint for XCom
Posted by GitBox <gi...@apache.org>.
mik-laj commented on pull request #9482:
URL: https://github.com/apache/airflow/pull/9482#issuecomment-653440785
@feluelle These API will not address this problem. I want to build an API for third systems, not an internal API. API for third-party systems should maintain backward compatibility and should provide access to the most important operations. Building an API that is too rich in features or provides too much information causes problems with maintaining backward compatibility.
I would not want this API to be only access to the database via HTTP, but it was an API that will easily allow you to build new applications on it, e.g. applications for monitoring multiple instances of Airflow and provide mult-tenant view.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] turbaszek commented on pull request #9482: Add CRUD endpoint for XCom
Posted by GitBox <gi...@apache.org>.
turbaszek commented on pull request #9482:
URL: https://github.com/apache/airflow/pull/9482#issuecomment-653484017
> We should allow setting TaskInstance state as we do allow that in the Webserver
True, I forgot about this. But still, this is very limited update option.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] feluelle commented on pull request #9482: Add CRUD endpoint for XCom
Posted by GitBox <gi...@apache.org>.
feluelle commented on pull request #9482:
URL: https://github.com/apache/airflow/pull/9482#issuecomment-653435834
@mik-laj I understand you point.
Hm.. I could imagine having multiple airflow instances which might want to share those xcom values with each other. WDYT?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] OmairK commented on a change in pull request #9482: Add CRUD endpoint for XCom
Posted by GitBox <gi...@apache.org>.
OmairK commented on a change in pull request #9482:
URL: https://github.com/apache/airflow/pull/9482#discussion_r445511936
##########
File path: airflow/api_connexion/endpoints/xcom_endpoint.py
##########
@@ -14,25 +14,47 @@
# KIND, either express or implied. See the License for the
# specific language governing permissions and limitations
# under the License.
-from flask import request
+from typing import List, Optional
+
+from flask import Response, request
+from marshmallow import ValidationError
from sqlalchemy import and_, func
from sqlalchemy.orm.session import Session
from airflow.api_connexion import parameters
-from airflow.api_connexion.exceptions import NotFound
+from airflow.api_connexion.exceptions import BadRequest, NotFound
from airflow.api_connexion.schemas.xcom_schema import (
XComCollection, XComCollectionItemSchema, XComCollectionSchema, xcom_collection_item_schema,
- xcom_collection_schema,
+ xcom_collection_schema, xcom_schema,
)
from airflow.models import DagRun as DR, XCom
from airflow.utils.session import provide_session
-def delete_xcom_entry():
+@provide_session
+def delete_xcom_entry(
+ dag_id: str,
+ dag_run_id: str,
+ task_id: str,
+ xcom_key: str,
+ session: Session
+) -> Response:
Review comment:
```suggestion
):
```
Mypy doesn't raise error it gets lost in [types](https://github.com/apache/airflow/pull/9329#discussion_r443198113)
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] ephraimbuddy commented on pull request #9482: Add CRUD endpoint for XCom
Posted by GitBox <gi...@apache.org>.
ephraimbuddy commented on pull request #9482:
URL: https://github.com/apache/airflow/pull/9482#issuecomment-649501803
@OmairK can I ask for review?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] feluelle commented on a change in pull request #9482: Add CRUD endpoint for XCom
Posted by GitBox <gi...@apache.org>.
feluelle commented on a change in pull request #9482:
URL: https://github.com/apache/airflow/pull/9482#discussion_r449462456
##########
File path: airflow/api_connexion/endpoints/xcom_endpoint.py
##########
@@ -89,15 +111,78 @@ def get_xcom_entry(
return xcom_collection_item_schema.dump(query_object)
-def patch_xcom_entry():
+@provide_session
+def patch_xcom_entry(
+ session: Session,
+ dag_id: str,
+ dag_run_id: str,
+ task_id: str,
+ xcom_key: str,
+ update_mask: Optional[List[str]] = None
+):
"""
Update an XCom entry
"""
- raise NotImplementedError("Not implemented yet.")
+ dag_run = session.query(DR).filter(DR.run_id == dag_run_id,
+ DR.dag_id == dag_id).first()
+ if not dag_run:
+ raise NotFound(f'DAGRun with dag_id:{dag_id} and run_id:{dag_run_id} not found')
+
+ xcom_entry = session.query(XCom).filter(XCom.dag_id == dag_id,
+ XCom.task_id == task_id,
+ XCom.key == xcom_key).first()
+
+ if not xcom_entry:
+ raise NotFound("XCom not found")
+ try:
+ body = xcom_schema.load(request.json, partial=("dag_id", "task_id",
+ "key", "execution_date"))
+ except ValidationError as err:
+ raise BadRequest(detail=err.messages.get('_schema', [str(err.messages)])[0])
+ data = body.data
+ # Check that other attributes are not being updated
+ for field in data:
+ if field != 'value' and data[field] != getattr(xcom_entry, field):
+ raise BadRequest(detail=f"'{field}' cannot be updated")
+ if update_mask:
+ update_mask = [i.strip() for i in update_mask]
+ data_ = {}
+ for field in update_mask:
+ if field in data and field != 'value':
+ raise BadRequest(detail=f"{field} cannot be updated")
+ elif field in data and field == 'value':
+ data_[field] = data[field]
+ else:
+ raise BadRequest(detail=f"Unknown field '{field}' in update_mask")
Review comment:
How about:
```python
if field not in data:
raise BadRequest(detail=f"Unknown field '{field}' in update_mask")
if field != 'value':
raise BadRequest(detail=f"{field} cannot be updated")
if field == 'value':
data_[field] = data[field]
```
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] ephraimbuddy commented on pull request #9482: Add CRUD endpoint for XCom
Posted by GitBox <gi...@apache.org>.
ephraimbuddy commented on pull request #9482:
URL: https://github.com/apache/airflow/pull/9482#issuecomment-649149124
Depends on : #9170
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] OmairK commented on a change in pull request #9482: Add CRUD endpoint for XCom
Posted by GitBox <gi...@apache.org>.
OmairK commented on a change in pull request #9482:
URL: https://github.com/apache/airflow/pull/9482#discussion_r445513453
##########
File path: airflow/api_connexion/endpoints/xcom_endpoint.py
##########
@@ -14,25 +14,47 @@
# KIND, either express or implied. See the License for the
# specific language governing permissions and limitations
# under the License.
-from flask import request
+from typing import List, Optional
+
+from flask import Response, request
+from marshmallow import ValidationError
from sqlalchemy import and_, func
from sqlalchemy.orm.session import Session
from airflow.api_connexion import parameters
-from airflow.api_connexion.exceptions import NotFound
+from airflow.api_connexion.exceptions import BadRequest, NotFound
from airflow.api_connexion.schemas.xcom_schema import (
XComCollection, XComCollectionItemSchema, XComCollectionSchema, xcom_collection_item_schema,
- xcom_collection_schema,
+ xcom_collection_schema, xcom_schema,
)
from airflow.models import DagRun as DR, XCom
from airflow.utils.session import provide_session
-def delete_xcom_entry():
+@provide_session
+def delete_xcom_entry(
+ dag_id: str,
+ dag_run_id: str,
+ task_id: str,
+ xcom_key: str,
+ session: Session
+) -> Response:
"""
Delete an XCom entry
"""
- raise NotImplementedError("Not implemented yet.")
+
+ dag_run = session.query(DR).filter(DR.run_id == dag_run_id,
+ DR.dag_id == dag_id).first()
+ if not dag_run:
+ raise NotFound(f'DAGRun with dag_id:{dag_id} and run_id:{dag_run_id} not found')
+
+ entry = session.query(XCom).filter(XCom.dag_id == dag_id,
+ XCom.task_id == task_id,
+ XCom.key == xcom_key).delete()
+ if not entry:
Review comment:
```suggestion
if session.query(XCom).filter(XCom.dag_id == dag_id,
XCom.task_id == task_id,
XCom.key == xcom_key).delete() == 0:
```
You could do something like this to make the code shorter delete returns the number of rows deleted
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] turbaszek commented on a change in pull request #9482: Add CRUD endpoint for XCom
Posted by GitBox <gi...@apache.org>.
turbaszek commented on a change in pull request #9482:
URL: https://github.com/apache/airflow/pull/9482#discussion_r449453250
##########
File path: airflow/api_connexion/endpoints/xcom_endpoint.py
##########
@@ -89,15 +111,78 @@ def get_xcom_entry(
return xcom_collection_item_schema.dump(query_object)
-def patch_xcom_entry():
+@provide_session
+def patch_xcom_entry(
+ session: Session,
+ dag_id: str,
+ dag_run_id: str,
+ task_id: str,
+ xcom_key: str,
+ update_mask: Optional[List[str]] = None
+):
"""
Update an XCom entry
"""
- raise NotImplementedError("Not implemented yet.")
+ dag_run = session.query(DR).filter(DR.run_id == dag_run_id,
+ DR.dag_id == dag_id).first()
+ if not dag_run:
+ raise NotFound(f'DAGRun with dag_id:{dag_id} and run_id:{dag_run_id} not found')
+
+ xcom_entry = session.query(XCom).filter(XCom.dag_id == dag_id,
+ XCom.task_id == task_id,
+ XCom.key == xcom_key).first()
+
+ if not xcom_entry:
+ raise NotFound("XCom not found")
+ try:
+ body = xcom_schema.load(request.json, partial=("dag_id", "task_id",
+ "key", "execution_date"))
+ except ValidationError as err:
+ raise BadRequest(detail=err.messages.get('_schema', [str(err.messages)])[0])
+ data = body.data
+ # Check that other attributes are not being updated
+ for field in data:
+ if field != 'value' and data[field] != getattr(xcom_entry, field):
+ raise BadRequest(detail=f"'{field}' cannot be updated")
+ if update_mask:
+ update_mask = [i.strip() for i in update_mask]
+ data_ = {}
+ for field in update_mask:
+ if field in data and field != 'value':
Review comment:
```suggestion
data_ = {}
for field in update_mask:
field = field.strip()
if field in data and field != 'value':
```
One loop only, WDYT?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] ephraimbuddy commented on a change in pull request #9482: Add CRUD endpoint for XCom
Posted by GitBox <gi...@apache.org>.
ephraimbuddy commented on a change in pull request #9482:
URL: https://github.com/apache/airflow/pull/9482#discussion_r449460858
##########
File path: airflow/api_connexion/endpoints/xcom_endpoint.py
##########
@@ -89,15 +111,78 @@ def get_xcom_entry(
return xcom_collection_item_schema.dump(query_object)
-def patch_xcom_entry():
+@provide_session
+def patch_xcom_entry(
+ session: Session,
+ dag_id: str,
+ dag_run_id: str,
+ task_id: str,
+ xcom_key: str,
+ update_mask: Optional[List[str]] = None
+):
"""
Update an XCom entry
"""
- raise NotImplementedError("Not implemented yet.")
+ dag_run = session.query(DR).filter(DR.run_id == dag_run_id,
+ DR.dag_id == dag_id).first()
+ if not dag_run:
+ raise NotFound(f'DAGRun with dag_id:{dag_id} and run_id:{dag_run_id} not found')
+
+ xcom_entry = session.query(XCom).filter(XCom.dag_id == dag_id,
+ XCom.task_id == task_id,
+ XCom.key == xcom_key).first()
+
+ if not xcom_entry:
+ raise NotFound("XCom not found")
+ try:
+ body = xcom_schema.load(request.json, partial=("dag_id", "task_id",
+ "key", "execution_date"))
+ except ValidationError as err:
+ raise BadRequest(detail=err.messages.get('_schema', [str(err.messages)])[0])
+ data = body.data
+ # Check that other attributes are not being updated
+ for field in data:
+ if field != 'value' and data[field] != getattr(xcom_entry, field):
+ raise BadRequest(detail=f"'{field}' cannot be updated")
+ if update_mask:
+ update_mask = [i.strip() for i in update_mask]
+ data_ = {}
+ for field in update_mask:
+ if field in data and field != 'value':
+ raise BadRequest(detail=f"{field} cannot be updated")
+ elif field in data and field == 'value':
+ data_[field] = data[field]
+ else:
+ raise BadRequest(detail=f"Unknown field '{field}' in update_mask")
Review comment:
Very nice :+1:
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] kaxil commented on pull request #9482: Add CRUD endpoint for XCom
Posted by GitBox <gi...@apache.org>.
kaxil commented on pull request #9482:
URL: https://github.com/apache/airflow/pull/9482#issuecomment-653468787
We should allow setting TaskInstance state as we do allow that in the Webserver
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org