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/16 12:12:35 UTC
[GitHub] [airflow] turbaszek opened a new pull request #9330: Add read-only Task endpoint
turbaszek opened a new pull request #9330:
URL: https://github.com/apache/airflow/pull/9330
Closes #8138
---
Make sure to mark the boxes below before creating PR: [x]
- [x] Description above provides context of the change
- [ ] Unit tests coverage for changes (not needed for documentation changes)
- [x] Target Github ISSUE in description if exists
- [x] Commits follow "[How to write a good git commit message](http://chris.beams.io/posts/git-commit/)"
- [x] Relevant documentation is updated including usage instructions.
- [x] 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] turbaszek commented on a change in pull request #9330: Add read-only Task endpoint
Posted by GitBox <gi...@apache.org>.
turbaszek commented on a change in pull request #9330:
URL: https://github.com/apache/airflow/pull/9330#discussion_r444818466
##########
File path: tests/cli/commands/test_dag_command.py
##########
@@ -59,8 +60,14 @@ class TestCliDags(unittest.TestCase):
@classmethod
def setUpClass(cls):
cls.dagbag = DagBag(include_examples=True)
+ DAG.bulk_sync_to_db([d[1] for d in cls.dagbag.dags.items()])
Review comment:
Done, thanks!
----------------------------------------------------------------
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 a change in pull request #9330: Add read-only Task endpoint
Posted by GitBox <gi...@apache.org>.
mik-laj commented on a change in pull request #9330:
URL: https://github.com/apache/airflow/pull/9330#discussion_r444288253
##########
File path: tests/cli/commands/test_dag_command.py
##########
@@ -59,8 +60,14 @@ class TestCliDags(unittest.TestCase):
@classmethod
def setUpClass(cls):
cls.dagbag = DagBag(include_examples=True)
+ DAG.bulk_sync_to_db([d[1] for d in cls.dagbag.dags.items()])
Review comment:
```suggestion
DAG.bulk_sync_to_db([d for d in cls.dagbag.dags.values()])
```
or
```suggestion
cls.dagbag.sync_to_db()
```
----------------------------------------------------------------
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 #9330: Add read-only Task endpoint
Posted by GitBox <gi...@apache.org>.
turbaszek commented on a change in pull request #9330:
URL: https://github.com/apache/airflow/pull/9330#discussion_r444241727
##########
File path: airflow/api_connexion/endpoints/task_endpoint.py
##########
@@ -14,20 +14,36 @@
# KIND, either express or implied. See the License for the
# specific language governing permissions and limitations
# under the License.
+from flask import current_app
-# TODO(mik-laj): We have to implement it.
-# Do you want to help? Please look at: https://github.com/apache/airflow/issues/8138
+from airflow import DAG
+from airflow.api_connexion.exceptions import NotFound
+from airflow.api_connexion.schemas.task_schema import TaskCollection, task_collection_schema, task_schema
+from airflow.exceptions import TaskNotFound
-def get_task():
+def get_task(dag_id, task_id):
"""
Get simplified representation of a task.
"""
- raise NotImplementedError("Not implemented yet.")
+ dag: DAG = current_app.dag_bag.get_dag(dag_id)
+ if not dag:
+ raise NotFound("DAG not found")
+ try:
+ task = dag.get_task(task_id=task_id)
+ except TaskNotFound:
+ raise NotFound("Task not found")
+ return task_schema.dump(task)
-def get_tasks():
+
+def get_tasks(dag_id):
"""
Get tasks for DAG
"""
- raise NotImplementedError("Not implemented yet.")
+ dag: DAG = current_app.dag_bag.get_dag(dag_id)
+ if not dag:
+ raise NotFound("DAG not found")
+ tasks = dag.tasks
+ task_collection = TaskCollection(tasks=tasks, total_entries=len(tasks))
Review comment:
Hm, I think here we return only tasks from single dag, so if I correctly understand the `total_entries` should be equal to the number of tasks in the requested DAG
----------------------------------------------------------------
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 a change in pull request #9330: Add read-only Task endpoint
Posted by GitBox <gi...@apache.org>.
mik-laj commented on a change in pull request #9330:
URL: https://github.com/apache/airflow/pull/9330#discussion_r440928429
##########
File path: airflow/api_connexion/endpoints/task_endpoint.py
##########
@@ -14,20 +14,42 @@
# KIND, either express or implied. See the License for the
# specific language governing permissions and limitations
# under the License.
+from airflow import DAG
+from airflow.api_connexion.exceptions import NotFound
+from airflow.api_connexion.schemas.task_schema import TaskCollection, task_collection_schema, task_schema
+from airflow.configuration import conf
+from airflow.exceptions import TaskNotFound
+from airflow.models import DagBag
-# TODO(mik-laj): We have to implement it.
-# Do you want to help? Please look at: https://github.com/apache/airflow/issues/8138
-
-def get_task():
+def get_task(dag_id, task_id):
"""
Get simplified representation of a task.
"""
- raise NotImplementedError("Not implemented yet.")
+ dag_bag = DagBag(
+ store_serialized_dags=conf.getboolean('core', 'store_serialized_dags'),
+ )
+ dag: DAG = dag_bag.get_dag(dag_id)
+ if not dag:
+ raise NotFound("DAG not found")
Review comment:
We should raise [NotFound](https://github.com/apache/airflow/blob/7c12a9d4e0b6c1e01fee6ab227a6e25b5aa5b157/airflow/api_connexion/exceptions.py#L20) exception. This class is correctly handled by connection and generates the response in the valid format.
----------------------------------------------------------------
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 #9330: Add read-only Task endpoint
Posted by GitBox <gi...@apache.org>.
turbaszek commented on a change in pull request #9330:
URL: https://github.com/apache/airflow/pull/9330#discussion_r442900275
##########
File path: tests/api_connexion/endpoints/test_dag_endpoint.py
##########
@@ -14,35 +14,109 @@
# KIND, either express or implied. See the License for the
# specific language governing permissions and limitations
# under the License.
+import os
import unittest
+from datetime import datetime
import pytest
+from airflow import DAG
+from airflow.models import DagBag
+from airflow.models.serialized_dag import SerializedDagModel
+from airflow.operators.dummy_operator import DummyOperator
from airflow.www import app
+from tests.test_utils.db import clear_db_dags, clear_db_runs, clear_db_serialized_dags
class TestDagEndpoint(unittest.TestCase):
+ dag_id = "test_dag"
+ task_id = "op1"
+
+ @staticmethod
+ def clean_db():
+ clear_db_runs()
+ clear_db_dags()
+ clear_db_serialized_dags()
+
@classmethod
def setUpClass(cls) -> None:
super().setUpClass()
cls.app = app.create_app(testing=True) # type:ignore
+ cls.app_serialized = app.create_app(testing=True) # type:ignore
+
+ with DAG(
+ cls.dag_id, start_date=datetime(2020, 6, 15), doc_md="details"
+ ) as dag:
+ DummyOperator(task_id=cls.task_id)
+
+ cls.dag = dag # type:ignore
+
+ dag_bag = DagBag(os.devnull, include_examples=False)
+ dag_bag.dags = {dag.dag_id: dag}
+ cls.app.dag_bag = dag_bag # type:ignore
+
+ dag_bag = DagBag(os.devnull, include_examples=False, store_serialized_dags=True)
+ cls.app_serialized.dag_bag = dag_bag # type:ignore
def setUp(self) -> None:
+ self.clean_db()
self.client = self.app.test_client() # type:ignore
+ self.client_serialized = self.app_serialized.test_client() # type:ignore
Review comment:
I am not sure if this is the right way but nothing else comes to my mind. What we have here are two testing clients one with single DAG dagbag and second one with empty one with serialization switched on
----------------------------------------------------------------
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 a change in pull request #9330: Add read-only Task endpoint
Posted by GitBox <gi...@apache.org>.
mik-laj commented on a change in pull request #9330:
URL: https://github.com/apache/airflow/pull/9330#discussion_r440835127
##########
File path: airflow/api_connexion/schemas/common_schema.py
##########
@@ -0,0 +1,145 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+import datetime
+import inspect
+import typing
+from functools import partial
+
+import marshmallow
+from dateutil import relativedelta
+from marshmallow import Schema, fields, validate
+from marshmallow_oneofschema import OneOfSchema
+
+from airflow.utils.weight_rule import WeightRule
+
+
+class CronExpression(typing.NamedTuple):
+ """Cron expression schema"""
+ value: str
+
+
+class TimeDeltaSchema(Schema):
+ """Time delta schema"""
+
+ objectType = fields.Constant("TimeDelta", dump_to="__type")
+ days = fields.Integer()
+ seconds = fields.Integer()
+ microsecond = fields.Integer()
+
+ @marshmallow.post_load
+ def make_time_delta(self, data, **kwargs):
+ """Create time delta based on data"""
+
+ if "objectType" in data:
+ del data["objectType"]
+ return datetime.timedelta(**data)
+
+
+class RelativeDeltaSchema(Schema):
+ """Relative delta schema"""
+
+ objectType = fields.Constant("RelativeDelta", dump_to="__type")
+ years = fields.Integer()
+ months = fields.Integer()
+ days = fields.Integer()
+ leapdays = fields.Integer()
+ hours = fields.Integer()
+ minutes = fields.Integer()
+ seconds = fields.Integer()
+ microseconds = fields.Integer()
+ year = fields.Integer()
+ month = fields.Integer()
+ day = fields.Integer()
+ hour = fields.Integer()
+ minute = fields.Integer()
+ second = fields.Integer()
+ microsecond = fields.Integer()
+
+ @marshmallow.post_load
+ def make_relative_delta(self, data, **kwargs):
+ """Create relative delta based on data"""
+
+ if "objectType" in data:
+ del data["objectType"]
+
+ return relativedelta.relativedelta(**data)
+
+
+class CronExpressionSchema(Schema):
+ """Cron expression schema"""
+
+ objectType = fields.Constant("CronExpression", dump_to="__type", required=True)
+ value = fields.String(required=True)
+
+ @marshmallow.post_load
+ def make_cron_expression(self, data, **kwargs):
+ """Create cron expression based on data"""
+ return CronExpression(data["value"])
+
+
+class ScheduleIntervalSchema(OneOfSchema):
+ """
+ Schedule interval.
+
+ It supports the following types:
+
+ * TimeDelta
+ * RelativeDelta
+ * CronExpression
+ """
+ type_field = "__type"
+ type_schemas = {
+ "TimeDelta": TimeDeltaSchema,
+ "RelativeDelta": RelativeDeltaSchema,
+ "CronExpression": CronExpressionSchema,
+ }
+
+ def _dump(self, obj, update_fields=True, **kwargs):
+ if isinstance(obj, str):
+ obj = CronExpression(obj)
+
+ return super()._dump(obj, update_fields=update_fields, **kwargs)
+
+ def get_obj_type(self, obj):
+ """Select schema based on object type"""
+ if isinstance(obj, datetime.timedelta):
+ return "TimeDelta"
+ elif isinstance(obj, relativedelta.relativedelta):
+ return "RelativeDelta"
+ elif isinstance(obj, CronExpression):
+ return "CronExpression"
+ else:
+ raise Exception("Unknown object type: {}".format(obj.__class__.__name__))
+
+
+ColorField = partial(fields.String, validate=validate.Regexp("^#[a-fA-F0-9]{3,6}$"))
Review comment:
Could you do less magic?
https://github.com/apache/airflow/blob/75bd2f180e4afb591b08ee5d2cdb9b85737258b8/airflow/api_connexion/schemas/enum_schemas.py
----------------------------------------------------------------
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 a change in pull request #9330: Add read-only Task endpoint
Posted by GitBox <gi...@apache.org>.
mik-laj commented on a change in pull request #9330:
URL: https://github.com/apache/airflow/pull/9330#discussion_r440837546
##########
File path: setup.py
##########
@@ -708,6 +708,8 @@ def is_package_excluded(package: str, exclusion_list: List[str]):
'lazy_object_proxy~=1.3',
'lockfile>=0.12.2',
'markdown>=2.5.2, <3.0',
+ 'marshmallow<3',
Review comment:
We don't suppport marsshmalllow yet. We have a dependency on FAB, which requires marshmallow 2.0 Support for 3.0 is WIP: https://github.com/dpgaspar/Flask-AppBuilder/pull/1382
----------------------------------------------------------------
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 #9330: Add read-only Task endpoint
Posted by GitBox <gi...@apache.org>.
turbaszek commented on a change in pull request #9330:
URL: https://github.com/apache/airflow/pull/9330#discussion_r444756955
##########
File path: tests/api_connexion/endpoints/test_dag_endpoint.py
##########
@@ -14,35 +14,109 @@
# KIND, either express or implied. See the License for the
# specific language governing permissions and limitations
# under the License.
+import os
import unittest
+from datetime import datetime
import pytest
+from airflow import DAG
+from airflow.models import DagBag
+from airflow.models.serialized_dag import SerializedDagModel
+from airflow.operators.dummy_operator import DummyOperator
from airflow.www import app
+from tests.test_utils.db import clear_db_dags, clear_db_runs, clear_db_serialized_dags
class TestDagEndpoint(unittest.TestCase):
+ dag_id = "test_dag"
+ task_id = "op1"
+
+ @staticmethod
+ def clean_db():
+ clear_db_runs()
+ clear_db_dags()
+ clear_db_serialized_dags()
+
@classmethod
def setUpClass(cls) -> None:
super().setUpClass()
cls.app = app.create_app(testing=True) # type:ignore
+ cls.app_serialized = app.create_app(testing=True) # type:ignore
+
+ with DAG(
+ cls.dag_id, start_date=datetime(2020, 6, 15), doc_md="details"
+ ) as dag:
+ DummyOperator(task_id=cls.task_id)
+
+ cls.dag = dag # type:ignore
+
+ dag_bag = DagBag(os.devnull, include_examples=False)
+ dag_bag.dags = {dag.dag_id: dag}
+ cls.app.dag_bag = dag_bag # type:ignore
+
+ dag_bag = DagBag(os.devnull, include_examples=False, store_serialized_dags=True)
+ cls.app_serialized.dag_bag = dag_bag # type:ignore
def setUp(self) -> None:
+ self.clean_db()
self.client = self.app.test_client() # type:ignore
+ self.client_serialized = self.app_serialized.test_client() # type:ignore
Review comment:
I was thinking about creating a new app in test with serialization, 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 commented on a change in pull request #9330: Add read-only Task endpoint
Posted by GitBox <gi...@apache.org>.
mik-laj commented on a change in pull request #9330:
URL: https://github.com/apache/airflow/pull/9330#discussion_r444283015
##########
File path: airflow/api_connexion/openapi/v1.yaml
##########
@@ -1964,13 +1972,14 @@ components:
__type: {type: string}
Review comment:
Should add other fields to required also?
----------------------------------------------------------------
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 a change in pull request #9330: Add read-only Task endpoint
Posted by GitBox <gi...@apache.org>.
mik-laj commented on a change in pull request #9330:
URL: https://github.com/apache/airflow/pull/9330#discussion_r444281700
##########
File path: airflow/api_connexion/endpoints/task_endpoint.py
##########
@@ -14,20 +14,36 @@
# KIND, either express or implied. See the License for the
# specific language governing permissions and limitations
# under the License.
+from flask import current_app
-# TODO(mik-laj): We have to implement it.
-# Do you want to help? Please look at: https://github.com/apache/airflow/issues/8138
+from airflow import DAG
+from airflow.api_connexion.exceptions import NotFound
+from airflow.api_connexion.schemas.task_schema import TaskCollection, task_collection_schema, task_schema
+from airflow.exceptions import TaskNotFound
-def get_task():
+def get_task(dag_id, task_id):
"""
Get simplified representation of a task.
"""
- raise NotImplementedError("Not implemented yet.")
+ dag: DAG = current_app.dag_bag.get_dag(dag_id)
+ if not dag:
+ raise NotFound("DAG not found")
+ try:
+ task = dag.get_task(task_id=task_id)
+ except TaskNotFound:
+ raise NotFound("Task not found")
+ return task_schema.dump(task)
-def get_tasks():
+
+def get_tasks(dag_id):
"""
Get tasks for DAG
"""
- raise NotImplementedError("Not implemented yet.")
+ dag: DAG = current_app.dag_bag.get_dag(dag_id)
+ if not dag:
+ raise NotFound("DAG not found")
+ tasks = dag.tasks
+ task_collection = TaskCollection(tasks=tasks, total_entries=len(tasks))
Review comment:
This endpoint has no pagination, because all objects must be loaded to be able to return a response. We don't use the database here, so we can't optimize it.
----------------------------------------------------------------
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 #9330: Add read-only Task endpoint
Posted by GitBox <gi...@apache.org>.
turbaszek commented on a change in pull request #9330:
URL: https://github.com/apache/airflow/pull/9330#discussion_r442898882
##########
File path: airflow/api_connexion/schemas/dag_schema.py
##########
@@ -0,0 +1,93 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+from typing import List, NamedTuple
+
+from marshmallow import Schema, fields
+from marshmallow_sqlalchemy import SQLAlchemySchema, auto_field
+
+from airflow.api_connexion.schemas.common_schema import TimeDeltaSchema, TimezoneField
+from airflow.models.dag import DagModel, DagTag
+
+
+class DagTagSchema(SQLAlchemySchema):
+ """Dag Tag schema"""
+ class Meta:
+ """Meta"""
+
+ model = DagTag
+
+ name = auto_field()
+
+
+class DAGSchema(SQLAlchemySchema):
+ """DAG schema"""
+
+ class Meta:
+ """Meta"""
+
+ model = DagModel
+
+ dag_id = auto_field(dump_only=True)
+ root_dag_id = auto_field(dump_only=True)
+ is_paused = auto_field(dump_only=True)
+ is_subdag = auto_field(dump_only=True)
+ fileloc = auto_field(dump_only=True)
+ owners = fields.Method("get_owners", dump_only=True)
+ description = auto_field(dump_only=True)
+ # schedule_interval = fields.Nested(ScheduleIntervalSchema, dump_only=True)
Review comment:
Had to figure what's wrong:
```
FAILED [100%][2020-06-19 16:27:55,965] {validation.py:230} ERROR - http://localhost/api/v1/dags/test_dag/details validation error: {'__type': 'TimeDelta', 'days': 1, 'seconds': 0} is valid under each of {'x-scope': ['', '#/components/schemas/DAGCollection', '#/components/schemas/DAG', '#/components/schemas/ScheduleInterval'], 'type': 'object', 'required': ['__type'], 'properties': {'__type': {'type': 'string'}, 'years': {'type': 'integer'}, 'months': {'type': 'integer'}, 'days': {'type': 'integer'}, 'leapdays': {'type': 'integer'}, 'hours': {'type': 'integer'}, 'minutes': {'type': 'integer'}, 'seconds': {'type': 'integer'}, 'microseconds': {'type': 'integer'}, 'year': {'type': 'integer'}, 'month': {'type': 'integer'}, 'day': {'type': 'integer'}, 'hour': {'type': 'integer'}, 'minute': {'type': 'integer'}, 'second': {'type': 'integer'}, 'microsecond': {'type': 'integer'}}}, {'x-scope': ['', '#/components/schemas/DAGCollection', '#/components/schemas/DAG', '#/components/schemas/ScheduleInterval'], 'type': 'object', 'required': ['__type'], 'properties': {'__type': {'type': 'string'}, 'value': {'type': 'string'}}}, {'x-scope': ['', '#/components/schemas/DAGCollection', '#/components/schemas/DAG', '#/components/schemas/ScheduleInterval'], 'type': 'object', 'required': ['__type'], 'properties': {'__type': {'type': 'string'}, 'days': {'type': 'integer'}, 'seconds': {'type': 'integer'}, 'microsecond': {'type': 'integer'}}}
Failed validating 'oneOf' in schema['allOf'][0]['properties']['schedule_interval']:
{'discriminator': {'propertyName': '__type'},
'oneOf': [{'properties': {'__type': {'type': 'string'},
'days': {'type': 'integer'},
'microsecond': {'type': 'integer'},
'seconds': {'type': 'integer'}},
'required': ['__type'],
'type': 'object',
'x-scope': ['',
'#/components/schemas/DAGCollection',
'#/components/schemas/DAG',
'#/components/schemas/ScheduleInterval']},
{'properties': {'__type': {'type': 'string'},
'day': {'type': 'integer'},
'days': {'type': 'integer'},
'hour': {'type': 'integer'},
'hours': {'type': 'integer'},
'leapdays': {'type': 'integer'},
'microsecond': {'type': 'integer'},
'microseconds': {'type': 'integer'},
'minute': {'type': 'integer'},
'minutes': {'type': 'integer'},
'month': {'type': 'integer'},
'months': {'type': 'integer'},
'second': {'type': 'integer'},
'seconds': {'type': 'integer'},
'year': {'type': 'integer'},
'years': {'type': 'integer'}},
'required': ['__type'],
'type': 'object',
'x-scope': ['',
'#/components/schemas/DAGCollection',
'#/components/schemas/DAG',
'#/components/schemas/ScheduleInterval']},
{'properties': {'__type': {'type': 'string'},
'value': {'type': 'string'}},
'required': ['__type'],
'type': 'object',
'x-scope': ['',
'#/components/schemas/DAGCollection',
'#/components/schemas/DAG',
'#/components/schemas/ScheduleInterval']}],
'readOnly': True,
'x-scope': ['',
'#/components/schemas/DAGCollection',
'#/components/schemas/DAG']}
On instance['schedule_interval']:
{'__type': 'TimeDelta', 'days': 1, 'seconds': 0}
```
----------------------------------------------------------------
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 a change in pull request #9330: Add read-only Task endpoint
Posted by GitBox <gi...@apache.org>.
mik-laj commented on a change in pull request #9330:
URL: https://github.com/apache/airflow/pull/9330#discussion_r440929154
##########
File path: airflow/api_connexion/openapi/v1.yaml
##########
@@ -1337,7 +1335,6 @@ components:
extra:
type: string
readOnly: true
- nullable: true
Review comment:
?
----------------------------------------------------------------
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 merged pull request #9330: Add read-only Task endpoint
Posted by GitBox <gi...@apache.org>.
turbaszek merged pull request #9330:
URL: https://github.com/apache/airflow/pull/9330
----------------------------------------------------------------
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 #9330: Add read-only Task endpoint
Posted by GitBox <gi...@apache.org>.
turbaszek commented on pull request #9330:
URL: https://github.com/apache/airflow/pull/9330#issuecomment-648079790
@ephraimbuddy would you mind taking a look?
----------------------------------------------------------------
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 #9330: Add read-only Task endpoint
Posted by GitBox <gi...@apache.org>.
turbaszek commented on a change in pull request #9330:
URL: https://github.com/apache/airflow/pull/9330#discussion_r444818356
##########
File path: airflow/api_connexion/openapi/v1.yaml
##########
@@ -1964,13 +1972,14 @@ components:
__type: {type: string}
days: {type: integer}
seconds: {type: integer}
- microsecond: {type: integer}
+ microseconds: {type: integer}
RelativeDelta:
# TODO: Why we need these fields?
type: object
required:
- __type
+ - years
Review comment:
Done
##########
File path: airflow/api_connexion/openapi/v1.yaml
##########
@@ -1964,13 +1972,14 @@ components:
__type: {type: string}
Review comment:
Done
----------------------------------------------------------------
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 a change in pull request #9330: Add read-only Task endpoint
Posted by GitBox <gi...@apache.org>.
mik-laj commented on a change in pull request #9330:
URL: https://github.com/apache/airflow/pull/9330#discussion_r444285880
##########
File path: tests/api_connexion/endpoints/test_dag_endpoint.py
##########
@@ -14,35 +14,109 @@
# KIND, either express or implied. See the License for the
# specific language governing permissions and limitations
# under the License.
+import os
import unittest
+from datetime import datetime
import pytest
+from airflow import DAG
+from airflow.models import DagBag
+from airflow.models.serialized_dag import SerializedDagModel
+from airflow.operators.dummy_operator import DummyOperator
from airflow.www import app
+from tests.test_utils.db import clear_db_dags, clear_db_runs, clear_db_serialized_dags
class TestDagEndpoint(unittest.TestCase):
+ dag_id = "test_dag"
+ task_id = "op1"
+
+ @staticmethod
+ def clean_db():
+ clear_db_runs()
+ clear_db_dags()
+ clear_db_serialized_dags()
+
@classmethod
def setUpClass(cls) -> None:
super().setUpClass()
cls.app = app.create_app(testing=True) # type:ignore
+ cls.app_serialized = app.create_app(testing=True) # type:ignore
+
+ with DAG(
+ cls.dag_id, start_date=datetime(2020, 6, 15), doc_md="details"
+ ) as dag:
+ DummyOperator(task_id=cls.task_id)
+
+ cls.dag = dag # type:ignore
+
+ dag_bag = DagBag(os.devnull, include_examples=False)
+ dag_bag.dags = {dag.dag_id: dag}
+ cls.app.dag_bag = dag_bag # type:ignore
+
+ dag_bag = DagBag(os.devnull, include_examples=False, store_serialized_dags=True)
+ cls.app_serialized.dag_bag = dag_bag # type:ignore
def setUp(self) -> None:
+ self.clean_db()
self.client = self.app.test_client() # type:ignore
+ self.client_serialized = self.app_serialized.test_client() # type:ignore
Review comment:
Maybe just give up this base class and divide the whole into two classes?
----------------------------------------------------------------
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 #9330: Add read-only Task endpoint
Posted by GitBox <gi...@apache.org>.
turbaszek commented on a change in pull request #9330:
URL: https://github.com/apache/airflow/pull/9330#discussion_r443627142
##########
File path: airflow/api_connexion/schemas/dag_schema.py
##########
@@ -0,0 +1,93 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+from typing import List, NamedTuple
+
+from marshmallow import Schema, fields
+from marshmallow_sqlalchemy import SQLAlchemySchema, auto_field
+
+from airflow.api_connexion.schemas.common_schema import TimeDeltaSchema, TimezoneField
+from airflow.models.dag import DagModel, DagTag
+
+
+class DagTagSchema(SQLAlchemySchema):
+ """Dag Tag schema"""
+ class Meta:
+ """Meta"""
+
+ model = DagTag
+
+ name = auto_field()
+
+
+class DAGSchema(SQLAlchemySchema):
+ """DAG schema"""
+
+ class Meta:
+ """Meta"""
+
+ model = DagModel
+
+ dag_id = auto_field(dump_only=True)
+ root_dag_id = auto_field(dump_only=True)
+ is_paused = auto_field(dump_only=True)
+ is_subdag = auto_field(dump_only=True)
+ fileloc = auto_field(dump_only=True)
+ owners = fields.Method("get_owners", dump_only=True)
+ description = auto_field(dump_only=True)
+ # schedule_interval = fields.Nested(ScheduleIntervalSchema, dump_only=True)
Review comment:
The problem seems to be solved by adding more than `__type` to `required` fields.
----------------------------------------------------------------
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 #9330: Add read-only Task endpoint
Posted by GitBox <gi...@apache.org>.
turbaszek commented on a change in pull request #9330:
URL: https://github.com/apache/airflow/pull/9330#discussion_r440963511
##########
File path: airflow/api_connexion/openapi/v1.yaml
##########
@@ -1337,7 +1335,6 @@ components:
extra:
type: string
readOnly: true
- nullable: true
Review comment:
Mistake, fixed
----------------------------------------------------------------
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 #9330: Add read-only Task endpoint
Posted by GitBox <gi...@apache.org>.
ephraimbuddy commented on a change in pull request #9330:
URL: https://github.com/apache/airflow/pull/9330#discussion_r444236641
##########
File path: airflow/api_connexion/endpoints/task_endpoint.py
##########
@@ -14,20 +14,36 @@
# KIND, either express or implied. See the License for the
# specific language governing permissions and limitations
# under the License.
+from flask import current_app
-# TODO(mik-laj): We have to implement it.
-# Do you want to help? Please look at: https://github.com/apache/airflow/issues/8138
+from airflow import DAG
+from airflow.api_connexion.exceptions import NotFound
+from airflow.api_connexion.schemas.task_schema import TaskCollection, task_collection_schema, task_schema
+from airflow.exceptions import TaskNotFound
-def get_task():
+def get_task(dag_id, task_id):
"""
Get simplified representation of a task.
"""
- raise NotImplementedError("Not implemented yet.")
+ dag: DAG = current_app.dag_bag.get_dag(dag_id)
+ if not dag:
+ raise NotFound("DAG not found")
+ try:
+ task = dag.get_task(task_id=task_id)
+ except TaskNotFound:
+ raise NotFound("Task not found")
+ return task_schema.dump(task)
-def get_tasks():
+
+def get_tasks(dag_id):
"""
Get tasks for DAG
"""
- raise NotImplementedError("Not implemented yet.")
+ dag: DAG = current_app.dag_bag.get_dag(dag_id)
+ if not dag:
+ raise NotFound("DAG not found")
+ tasks = dag.tasks
+ task_collection = TaskCollection(tasks=tasks, total_entries=len(tasks))
Review comment:
The` total_entries` is not actually the total of query result. In this case now, It is the total of all tasks in the db.
----------------------------------------------------------------
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 #9330: Add read-only Task endpoint
Posted by GitBox <gi...@apache.org>.
turbaszek commented on pull request #9330:
URL: https://github.com/apache/airflow/pull/9330#issuecomment-646060590
Depends on https://github.com/apache/airflow/pull/9380
----------------------------------------------------------------
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 a change in pull request #9330: Add read-only Task endpoint
Posted by GitBox <gi...@apache.org>.
mik-laj commented on a change in pull request #9330:
URL: https://github.com/apache/airflow/pull/9330#discussion_r444282120
##########
File path: airflow/api_connexion/openapi/v1.yaml
##########
@@ -1964,13 +1972,14 @@ components:
__type: {type: string}
days: {type: integer}
seconds: {type: integer}
- microsecond: {type: integer}
+ microseconds: {type: integer}
RelativeDelta:
# TODO: Why we need these fields?
type: object
required:
- __type
+ - years
Review comment:
Should we add other fields also?
----------------------------------------------------------------
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