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/30 20:59:45 UTC
[GitHub] [airflow] mik-laj opened a new pull request #9597: [WIP] Add read-only endpoints for task instances
mik-laj opened a new pull request #9597:
URL: https://github.com/apache/airflow/pull/9597
---
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] kaxil merged pull request #9597: Add endpoints for task instances
Posted by GitBox <gi...@apache.org>.
kaxil merged pull request #9597:
URL: https://github.com/apache/airflow/pull/9597
----------------------------------------------------------------
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 #9597: [WIP] Add read-only endpoints for task instances
Posted by GitBox <gi...@apache.org>.
mik-laj commented on pull request #9597:
URL: https://github.com/apache/airflow/pull/9597#issuecomment-706051780
Everything works. I am merging all commits into one commit on my branch and doing rebase now
----------------------------------------------------------------
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 #9597: [WIP] Add read-only endpoints for task instances
Posted by GitBox <gi...@apache.org>.
mik-laj commented on a change in pull request #9597:
URL: https://github.com/apache/airflow/pull/9597#discussion_r448588204
##########
File path: airflow/api_connexion/endpoints/task_instance_endpoint.py
##########
@@ -14,23 +14,110 @@
# KIND, either express or implied. See the License for the
# specific language governing permissions and limitations
# under the License.
+from typing import Any, List, Optional, Tuple
-# TODO(mik-laj): We have to implement it.
-# Do you want to help? Please look at: https://github.com/apache/airflow/issues/8132
+from sqlalchemy import and_, func, or_
+from airflow.api_connexion.exceptions import NotFound
+from airflow.api_connexion.parameters import format_datetime, format_parameters
+from airflow.api_connexion.schemas.task_instance_schema import (
+ TaskInstanceCollection, task_instance_collection_schema, task_instance_schema,
+)
+from airflow.models.dagrun import DagRun as DR
+from airflow.models.taskinstance import TaskInstance as TI
+from airflow.utils.session import provide_session
+
+
+@provide_session
+def get_task_instance(dag_id: str, dag_run_id: str, task_id: str, session=None):
+ query = (
+ session.query(TI)
+ .filter(TI.dag_id == dag_id)
+ .join(DR, and_(TI.dag_id == DR.dag_id, TI.execution_date == DR.execution_date))
+ .filter(DR.run_id == dag_run_id)
+ .filter(TI.task_id == task_id)
+ )
+
+ task_instance = query.one_or_none()
+
+ if task_instance is None:
+ raise NotFound("Task instance not found")
+
+ return task_instance_schema.dump(task_instance)
-def get_task_instance():
- """
- Get a task instance
- """
- raise NotImplementedError("Not implemented yet.")
+def _apply_array_filter(query, key, values):
+ if values is not None:
+ query = query.filter(or_(*[key == v for v in values]))
+ return query
-def get_task_instances():
+
+def _apply_range_filter(query, key, value_range: Tuple[Any, Any]):
+ gte_value, lte_value = value_range
+ if gte_value is not None:
+ query = query.filter(key >= gte_value)
+ if lte_value is not None:
+ query = query.filter(key <= lte_value)
+ return query
+
+
+@format_parameters(
+ {
+ 'start_date_gte': format_datetime,
+ 'start_date_lte': format_datetime,
+ 'execution_date_gte': format_datetime,
+ 'execution_date_lte': format_datetime,
+ 'end_date_gte': format_datetime,
+ 'end_date_lte': format_datetime,
+ }
+)
+@provide_session
+def get_task_instances(
+ limit: int,
+ dag_id: Optional[str] = None,
+ dag_run_id: Optional[str] = None,
+ execution_date_gte: Optional[str] = None,
+ execution_date_lte: Optional[str] = None,
+ start_date_gte: Optional[str] = None,
+ start_date_lte: Optional[str] = None,
+ end_date_gte: Optional[str] = None,
+ end_date_lte: Optional[str] = None,
+ duration_gte: Optional[float] = None,
+ duration_lte: Optional[float] = None,
+ state: Optional[str] = None,
+ pool: Optional[List[str]] = None,
+ queue: Optional[List[str]] = None,
+ offset: Optional[int] = None,
+ session=None,
+):
"""
- Get list of task instances of DAG.
+ Get list of a task instances
"""
- raise NotImplementedError("Not implemented yet.")
+ query = session.query(TI)
+
+ if dag_id is not None:
+ query = query.filter(TI.dag_id == dag_id)
+ if dag_run_id is not None:
+ query = query.join(DR, and_(TI.dag_id == DR.dag_id, TI.execution_date == DR.execution_date))
+ query = query.filter(DR.run_id == dag_run_id)
+
+ query = _apply_range_filter(
+ query, key=DR.execution_date, value_range=(execution_date_gte, execution_date_lte)
+ )
+ query = _apply_range_filter(query, key=DR.start_date, value_range=(start_date_gte, start_date_lte))
+ query = _apply_range_filter(query, key=DR.end_date, value_range=(end_date_gte, end_date_lte))
+ query = _apply_range_filter(query, key=DR.end_date, value_range=(end_date_gte, end_date_lte))
Review comment:
Good point. I will remove 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] mik-laj commented on a change in pull request #9597: [WIP] Add read-only endpoints for task instances
Posted by GitBox <gi...@apache.org>.
mik-laj commented on a change in pull request #9597:
URL: https://github.com/apache/airflow/pull/9597#discussion_r448586439
##########
File path: airflow/api_connexion/endpoints/task_instance_endpoint.py
##########
@@ -14,23 +14,110 @@
# KIND, either express or implied. See the License for the
# specific language governing permissions and limitations
# under the License.
+from typing import Any, List, Optional, Tuple
-# TODO(mik-laj): We have to implement it.
-# Do you want to help? Please look at: https://github.com/apache/airflow/issues/8132
+from sqlalchemy import and_, func, or_
+from airflow.api_connexion.exceptions import NotFound
+from airflow.api_connexion.parameters import format_datetime, format_parameters
+from airflow.api_connexion.schemas.task_instance_schema import (
+ TaskInstanceCollection, task_instance_collection_schema, task_instance_schema,
+)
+from airflow.models.dagrun import DagRun as DR
+from airflow.models.taskinstance import TaskInstance as TI
+from airflow.utils.session import provide_session
+
+
+@provide_session
+def get_task_instance(dag_id: str, dag_run_id: str, task_id: str, session=None):
Review comment:
Inside Airflow, we use execution_date as the primary identifier. However, we voted to use dag_run_id in the API
https://lists.apache.org/thread.html/rd4be3829627dcef8b40314c62c041f460992786f3bfcc634d25a6664%40%3Cdev.airflow.apache.org%3E
----------------------------------------------------------------
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 #9597: [WIP] Add read-only endpoints for task instances
Posted by GitBox <gi...@apache.org>.
OmairK commented on a change in pull request #9597:
URL: https://github.com/apache/airflow/pull/9597#discussion_r479585913
##########
File path: airflow/api_connexion/endpoints/task_instance_endpoint.py
##########
@@ -14,39 +14,234 @@
# KIND, either express or implied. See the License for the
# specific language governing permissions and limitations
# under the License.
+from typing import Any, List, Optional, Tuple
+
+from flask import request
+from marshmallow import ValidationError
+from sqlalchemy import and_, func, or_
-# TODO(mik-laj): We have to implement it.
-# Do you want to help? Please look at: https://github.com/apache/airflow/issues/8132
from airflow.api_connexion import security
+from airflow.api_connexion.exceptions import NotFound, BadRequest
+from airflow.api_connexion.parameters import format_datetime, format_parameters
+from airflow.api_connexion.schemas.task_instance_schema import (
+ clear_task_instance_form,
+ TaskInstanceCollection,
+ task_instance_collection_schema,
+ task_instance_schema,
+ task_instance_batch_form,
+ task_instance_reference_collection_schema,
+ TaskInstanceReferenceCollection,
+)
+from airflow.configuration import conf
+from airflow.models.dagrun import DagRun as DR
+from airflow.models.taskinstance import clear_task_instances, TaskInstance as TI
+from airflow.models.dag import DagModel
+from airflow.models.dagbag import DagBag
+from airflow.models import SlaMiss
+from airflow.utils.state import State
+from airflow.utils.session import provide_session
@security.requires_authentication
-def get_task_instance():
- """
- Get a task instance
- """
- raise NotImplementedError("Not implemented yet.")
+@provide_session
+def get_task_instance(dag_id: str, dag_run_id: str, task_id: str, session=None):
+ """Get task instance"""
+ query = (
+ session.query(TI)
+ .filter(TI.dag_id == dag_id)
+ .join(DR, and_(TI.dag_id == DR.dag_id, TI.execution_date == DR.execution_date))
+ .filter(DR.run_id == dag_run_id)
+ .filter(TI.task_id == task_id)
+ .join(
+ SlaMiss,
+ and_(
+ SlaMiss.dag_id == TI.dag_id,
+ SlaMiss.execution_date == TI.execution_date,
+ SlaMiss.task_id == TI.task_id,
+ ),
+ isouter=True,
+ )
+ .add_entity(SlaMiss)
+ )
+ task_instance = query.one_or_none()
+
+ if task_instance is None:
+ raise NotFound("Task instance not found")
+
+ return task_instance_schema.dump(task_instance)
+
+
+def _apply_array_filter(query, key, values):
+ if values is not None:
+ query = query.filter(or_(*[key == v for v in values]))
+ return query
+
+
+def _apply_range_filter(query, key, value_range: Tuple[Any, Any]):
+ gte_value, lte_value = value_range
+ if gte_value is not None:
+ query = query.filter(key >= gte_value)
+ if lte_value is not None:
+ query = query.filter(key <= lte_value)
+ return query
+@format_parameters(
+ {
+ "execution_date_gte": format_datetime,
+ "execution_date_lte": format_datetime,
+ "start_date_gte": format_datetime,
+ "start_date_lte": format_datetime,
+ "end_date_gte": format_datetime,
+ "end_date_lte": format_datetime,
+ }
+)
@security.requires_authentication
-def get_task_instances():
+@provide_session
+def get_task_instances(
+ limit: int,
+ dag_id: Optional[str] = None,
+ dag_run_id: Optional[str] = None,
+ execution_date_gte: Optional[str] = None,
+ execution_date_lte: Optional[str] = None,
+ start_date_gte: Optional[str] = None,
+ start_date_lte: Optional[str] = None,
+ end_date_gte: Optional[str] = None,
+ end_date_lte: Optional[str] = None,
+ duration_gte: Optional[float] = None,
+ duration_lte: Optional[float] = None,
+ state: Optional[str] = None,
+ pool: Optional[List[str]] = None,
+ queue: Optional[List[str]] = None,
+ offset: Optional[int] = None,
+ session=None,
+): # pylint: disable=too-many-arguments
"""
- Get list of task instances of DAG.
+ Get list of task instances.
"""
- raise NotImplementedError("Not implemented yet.")
+ query = session.query(TI)
+
+ if dag_id != "~":
+ query = query.filter(TI.dag_id == dag_id)
+ if dag_run_id != "~":
+ query = query.join(DR, and_(TI.dag_id == DR.dag_id, TI.execution_date == DR.execution_date))
+ query = query.filter(DR.run_id == dag_run_id)
+ query = _apply_range_filter(
+ query, key=DR.execution_date, value_range=(execution_date_gte, execution_date_lte),
+ )
+ query = _apply_range_filter(query, key=TI.start_date, value_range=(start_date_gte, start_date_lte))
+ query = _apply_range_filter(query, key=TI.end_date, value_range=(end_date_gte, end_date_lte))
+ query = _apply_range_filter(query, key=TI.duration, value_range=(duration_gte, duration_lte))
+ query = _apply_array_filter(query, key=TI.state, values=state)
+ query = _apply_array_filter(query, key=TI.pool, values=pool)
+ query = _apply_array_filter(query, key=TI.queue, values=queue)
+
+ query = query.join(
+ SlaMiss,
+ and_(
+ SlaMiss.dag_id == TI.dag_id,
+ SlaMiss.task_id == TI.task_id,
+ SlaMiss.execution_date == TI.execution_date,
+ ),
+ isouter=True,
+ ).add_entity(SlaMiss)
+ task_instances = query.offset(offset).limit(limit).all()
+ total_entries = session.query(
+ func.count(func.distinct(TI.dag_id, TI.execution_date, TI.task_id))
+ ).scalar()
+ return task_instance_collection_schema.dump(
+ TaskInstanceCollection(task_instances=task_instances, total_entries=total_entries)
+ )
@security.requires_authentication
-def get_task_instances_batch():
+@provide_session
+def get_task_instances_batch(session=None):
"""
Get list of task instances.
"""
- raise NotImplementedError("Not implemented yet.")
+ body = request.get_json()
+ try:
+ data = task_instance_batch_form.load(body)
+ except ValidationError as err:
+ raise BadRequest(detail=str(err.messages))
+ query = session.query(TI)
+
+ query = _apply_array_filter(query, key=TI.dag_id, values=data["dag_ids"])
+ query = _apply_range_filter(
+ query, key=TI.execution_date, value_range=(data["execution_date_gte"], data["execution_date_lte"]),
+ )
+ query = _apply_range_filter(
+ query, key=TI.start_date, value_range=(data["start_date_gte"], data["start_date_lte"]),
+ )
+ query = _apply_range_filter(
+ query, key=TI.end_date, value_range=(data["end_date_gte"], data["end_date_lte"])
+ )
+ query = _apply_range_filter(
+ query, key=TI.duration, value_range=(data["duration_gte"], data["duration_lte"])
+ )
+ query = _apply_array_filter(query, key=TI.state, values=data["state"])
+ query = _apply_array_filter(query, key=TI.pool, values=data["pool"])
+ query = _apply_array_filter(query, key=TI.queue, values=data["queue"])
+
+ query = query.join(
+ SlaMiss,
+ and_(
+ SlaMiss.dag_id == TI.dag_id,
+ SlaMiss.task_id == TI.task_id,
+ SlaMiss.execution_date == TI.execution_date,
+ ),
+ isouter=True,
+ ).add_entity(SlaMiss)
Review comment:
@mik-laj I have dealt with the n +1 problem, but I am unsure if this query is the best way to go forward.
----------------------------------------------------------------
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 #9597: [WIP] Add read-only endpoints for task instances
Posted by GitBox <gi...@apache.org>.
OmairK commented on a change in pull request #9597:
URL: https://github.com/apache/airflow/pull/9597#discussion_r488171031
##########
File path: airflow/api_connexion/endpoints/task_instance_endpoint.py
##########
@@ -14,39 +14,234 @@
# KIND, either express or implied. See the License for the
# specific language governing permissions and limitations
# under the License.
+from typing import Any, List, Optional, Tuple
+
+from flask import request
+from marshmallow import ValidationError
+from sqlalchemy import and_, func, or_
-# TODO(mik-laj): We have to implement it.
-# Do you want to help? Please look at: https://github.com/apache/airflow/issues/8132
from airflow.api_connexion import security
+from airflow.api_connexion.exceptions import NotFound, BadRequest
+from airflow.api_connexion.parameters import format_datetime, format_parameters
+from airflow.api_connexion.schemas.task_instance_schema import (
+ clear_task_instance_form,
+ TaskInstanceCollection,
+ task_instance_collection_schema,
+ task_instance_schema,
+ task_instance_batch_form,
+ task_instance_reference_collection_schema,
+ TaskInstanceReferenceCollection,
+)
+from airflow.configuration import conf
+from airflow.models.dagrun import DagRun as DR
+from airflow.models.taskinstance import clear_task_instances, TaskInstance as TI
+from airflow.models.dag import DagModel
+from airflow.models.dagbag import DagBag
+from airflow.models import SlaMiss
+from airflow.utils.state import State
+from airflow.utils.session import provide_session
@security.requires_authentication
-def get_task_instance():
- """
- Get a task instance
- """
- raise NotImplementedError("Not implemented yet.")
+@provide_session
+def get_task_instance(dag_id: str, dag_run_id: str, task_id: str, session=None):
+ """Get task instance"""
+ query = (
+ session.query(TI)
+ .filter(TI.dag_id == dag_id)
+ .join(DR, and_(TI.dag_id == DR.dag_id, TI.execution_date == DR.execution_date))
+ .filter(DR.run_id == dag_run_id)
+ .filter(TI.task_id == task_id)
+ .join(
+ SlaMiss,
+ and_(
+ SlaMiss.dag_id == TI.dag_id,
+ SlaMiss.execution_date == TI.execution_date,
+ SlaMiss.task_id == TI.task_id,
+ ),
+ isouter=True,
+ )
+ .add_entity(SlaMiss)
+ )
+ task_instance = query.one_or_none()
+
+ if task_instance is None:
+ raise NotFound("Task instance not found")
+
+ return task_instance_schema.dump(task_instance)
+
+
+def _apply_array_filter(query, key, values):
+ if values is not None:
+ query = query.filter(or_(*[key == v for v in values]))
+ return query
+
+
+def _apply_range_filter(query, key, value_range: Tuple[Any, Any]):
+ gte_value, lte_value = value_range
+ if gte_value is not None:
+ query = query.filter(key >= gte_value)
+ if lte_value is not None:
+ query = query.filter(key <= lte_value)
+ return query
+@format_parameters(
+ {
+ "execution_date_gte": format_datetime,
+ "execution_date_lte": format_datetime,
+ "start_date_gte": format_datetime,
+ "start_date_lte": format_datetime,
+ "end_date_gte": format_datetime,
+ "end_date_lte": format_datetime,
+ }
+)
@security.requires_authentication
-def get_task_instances():
+@provide_session
+def get_task_instances(
+ limit: int,
+ dag_id: Optional[str] = None,
+ dag_run_id: Optional[str] = None,
+ execution_date_gte: Optional[str] = None,
+ execution_date_lte: Optional[str] = None,
+ start_date_gte: Optional[str] = None,
+ start_date_lte: Optional[str] = None,
+ end_date_gte: Optional[str] = None,
+ end_date_lte: Optional[str] = None,
+ duration_gte: Optional[float] = None,
+ duration_lte: Optional[float] = None,
+ state: Optional[str] = None,
+ pool: Optional[List[str]] = None,
+ queue: Optional[List[str]] = None,
+ offset: Optional[int] = None,
+ session=None,
+): # pylint: disable=too-many-arguments
"""
- Get list of task instances of DAG.
+ Get list of task instances.
"""
- raise NotImplementedError("Not implemented yet.")
+ query = session.query(TI)
+
+ if dag_id != "~":
+ query = query.filter(TI.dag_id == dag_id)
+ if dag_run_id != "~":
+ query = query.join(DR, and_(TI.dag_id == DR.dag_id, TI.execution_date == DR.execution_date))
+ query = query.filter(DR.run_id == dag_run_id)
+ query = _apply_range_filter(
+ query, key=DR.execution_date, value_range=(execution_date_gte, execution_date_lte),
+ )
+ query = _apply_range_filter(query, key=TI.start_date, value_range=(start_date_gte, start_date_lte))
+ query = _apply_range_filter(query, key=TI.end_date, value_range=(end_date_gte, end_date_lte))
+ query = _apply_range_filter(query, key=TI.duration, value_range=(duration_gte, duration_lte))
+ query = _apply_array_filter(query, key=TI.state, values=state)
+ query = _apply_array_filter(query, key=TI.pool, values=pool)
+ query = _apply_array_filter(query, key=TI.queue, values=queue)
+
+ query = query.join(
+ SlaMiss,
+ and_(
+ SlaMiss.dag_id == TI.dag_id,
+ SlaMiss.task_id == TI.task_id,
+ SlaMiss.execution_date == TI.execution_date,
+ ),
+ isouter=True,
+ ).add_entity(SlaMiss)
+ task_instances = query.offset(offset).limit(limit).all()
+ total_entries = session.query(
+ func.count(func.distinct(TI.dag_id, TI.execution_date, TI.task_id))
+ ).scalar()
+ return task_instance_collection_schema.dump(
+ TaskInstanceCollection(task_instances=task_instances, total_entries=total_entries)
+ )
@security.requires_authentication
-def get_task_instances_batch():
+@provide_session
+def get_task_instances_batch(session=None):
"""
Get list of task instances.
"""
- raise NotImplementedError("Not implemented yet.")
+ body = request.get_json()
+ try:
+ data = task_instance_batch_form.load(body)
+ except ValidationError as err:
+ raise BadRequest(detail=str(err.messages))
+ query = session.query(TI)
+
+ query = _apply_array_filter(query, key=TI.dag_id, values=data["dag_ids"])
+ query = _apply_range_filter(
+ query, key=TI.execution_date, value_range=(data["execution_date_gte"], data["execution_date_lte"]),
+ )
+ query = _apply_range_filter(
+ query, key=TI.start_date, value_range=(data["start_date_gte"], data["start_date_lte"]),
+ )
+ query = _apply_range_filter(
+ query, key=TI.end_date, value_range=(data["end_date_gte"], data["end_date_lte"])
+ )
+ query = _apply_range_filter(
+ query, key=TI.duration, value_range=(data["duration_gte"], data["duration_lte"])
+ )
+ query = _apply_array_filter(query, key=TI.state, values=data["state"])
+ query = _apply_array_filter(query, key=TI.pool, values=data["pool"])
+ query = _apply_array_filter(query, key=TI.queue, values=data["queue"])
+
+ query = query.join(
+ SlaMiss,
+ and_(
+ SlaMiss.dag_id == TI.dag_id,
+ SlaMiss.task_id == TI.task_id,
+ SlaMiss.execution_date == TI.execution_date,
+ ),
+ isouter=True,
+ ).add_entity(SlaMiss)
Review comment:
@mik-laj is there a better approach that I should work on or is this fine?
----------------------------------------------------------------
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 #9597: Add endpoints for task instances
Posted by GitBox <gi...@apache.org>.
kaxil commented on a change in pull request #9597:
URL: https://github.com/apache/airflow/pull/9597#discussion_r503504005
##########
File path: airflow/api_connexion/endpoints/task_instance_endpoint.py
##########
@@ -14,46 +14,260 @@
# KIND, either express or implied. See the License for the
# specific language governing permissions and limitations
# under the License.
+from typing import Any, List, Optional, Tuple
+
+from flask import request
+from marshmallow import ValidationError
+from sqlalchemy import and_, func
-# TODO(mik-laj): We have to implement it.
-# Do you want to help? Please look at: https://github.com/apache/airflow/issues/8132
from airflow.api_connexion import security
+from airflow.api_connexion.exceptions import NotFound, BadRequest
+from airflow.api_connexion.parameters import format_datetime, format_parameters
+from airflow.api_connexion.schemas.task_instance_schema import (
+ clear_task_instance_form,
+ TaskInstanceCollection,
+ task_instance_collection_schema,
+ task_instance_schema,
+ task_instance_batch_form,
+ task_instance_reference_collection_schema,
+ TaskInstanceReferenceCollection,
+)
+from airflow.configuration import conf
+from airflow.models.dagrun import DagRun as DR
+from airflow.models.taskinstance import clear_task_instances, TaskInstance as TI
+from airflow.models.dag import DagModel
+from airflow.models.dagbag import DagBag
+from airflow.models import SlaMiss
+from airflow.utils.state import State
+from airflow.utils.session import provide_session
-@security.requires_access([("can_read", "Dag"), ("can_read", "DagRun"), ("can_read", "Task")])
-def get_task_instance():
- """
- Get a task instance
- """
- raise NotImplementedError("Not implemented yet.")
+@security.requires_access(
+ [
+ ("can_read", "Dag"),
+ ("can_read", "DagRun"),
+ ('can_read', 'Task'),
+ ]
+)
+@provide_session
+def get_task_instance(dag_id: str, dag_run_id: str, task_id: str, session=None):
+ """Get task instance"""
+ query = (
+ session.query(TI)
+ .filter(TI.dag_id == dag_id)
+ .join(DR, and_(TI.dag_id == DR.dag_id, TI.execution_date == DR.execution_date))
+ .filter(DR.run_id == dag_run_id)
+ .filter(TI.task_id == task_id)
+ .join(
Review comment:
```suggestion
.outerjoin(
```
if we want to do an outerjoin instead of isouter=True
----------------------------------------------------------------
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 #9597: Add endpoints for task instances
Posted by GitBox <gi...@apache.org>.
mik-laj commented on a change in pull request #9597:
URL: https://github.com/apache/airflow/pull/9597#discussion_r502402326
##########
File path: airflow/api_connexion/endpoints/task_instance_endpoint.py
##########
@@ -14,39 +14,243 @@
# KIND, either express or implied. See the License for the
# specific language governing permissions and limitations
# under the License.
+from typing import Any, List, Optional, Tuple
+
+from flask import request
+from marshmallow import ValidationError
+from sqlalchemy import and_, func, or_
-# TODO(mik-laj): We have to implement it.
-# Do you want to help? Please look at: https://github.com/apache/airflow/issues/8132
from airflow.api_connexion import security
+from airflow.api_connexion.exceptions import NotFound, BadRequest
+from airflow.api_connexion.parameters import format_datetime, format_parameters
+from airflow.api_connexion.schemas.task_instance_schema import (
+ clear_task_instance_form,
+ TaskInstanceCollection,
+ task_instance_collection_schema,
+ task_instance_schema,
+ task_instance_batch_form,
+ task_instance_reference_collection_schema,
+ TaskInstanceReferenceCollection,
+)
+from airflow.configuration import conf
+from airflow.models.dagrun import DagRun as DR
+from airflow.models.taskinstance import clear_task_instances, TaskInstance as TI
+from airflow.models.dag import DagModel
+from airflow.models.dagbag import DagBag
+from airflow.models import SlaMiss
+from airflow.utils.state import State
+from airflow.utils.session import provide_session
@security.requires_authentication
-def get_task_instance():
- """
- Get a task instance
- """
- raise NotImplementedError("Not implemented yet.")
+@provide_session
+def get_task_instance(dag_id: str, dag_run_id: str, task_id: str, session=None):
+ """Get task instance"""
+ query = (
+ session.query(TI)
+ .filter(TI.dag_id == dag_id)
+ .join(DR, and_(TI.dag_id == DR.dag_id, TI.execution_date == DR.execution_date))
+ .filter(DR.run_id == dag_run_id)
+ .filter(TI.task_id == task_id)
+ .join(
+ SlaMiss,
+ and_(
+ SlaMiss.dag_id == TI.dag_id,
+ SlaMiss.execution_date == TI.execution_date,
+ SlaMiss.task_id == TI.task_id,
+ ),
+ isouter=True,
+ )
+ .add_entity(SlaMiss)
+ )
+ task_instance = query.one_or_none()
+ if task_instance is None:
+ raise NotFound("Task instance not found")
+
+ return task_instance_schema.dump(task_instance)
+
+
+def _apply_array_filter(query, key, values):
+ if values is not None:
+ query = query.filter(or_(*[key == v for v in values]))
+ return query
+
+
+def _apply_range_filter(query, key, value_range: Tuple[Any, Any]):
+ gte_value, lte_value = value_range
+ if gte_value is not None:
+ query = query.filter(key >= gte_value)
+ if lte_value is not None:
+ query = query.filter(key <= lte_value)
+ return query
+@format_parameters(
+ {
+ "execution_date_gte": format_datetime,
+ "execution_date_lte": format_datetime,
+ "start_date_gte": format_datetime,
+ "start_date_lte": format_datetime,
+ "end_date_gte": format_datetime,
+ "end_date_lte": format_datetime,
+ }
+)
@security.requires_authentication
-def get_task_instances():
+@provide_session
+def get_task_instances(
+ limit: int,
+ dag_id: Optional[str] = None,
+ dag_run_id: Optional[str] = None,
+ execution_date_gte: Optional[str] = None,
+ execution_date_lte: Optional[str] = None,
+ start_date_gte: Optional[str] = None,
+ start_date_lte: Optional[str] = None,
+ end_date_gte: Optional[str] = None,
+ end_date_lte: Optional[str] = None,
+ duration_gte: Optional[float] = None,
+ duration_lte: Optional[float] = None,
+ state: Optional[str] = None,
+ pool: Optional[List[str]] = None,
+ queue: Optional[List[str]] = None,
+ offset: Optional[int] = None,
+ session=None,
+): # pylint: disable=too-many-arguments
"""
- Get list of task instances of DAG.
+ Get list of task instances.
"""
- raise NotImplementedError("Not implemented yet.")
+ query = session.query(TI)
+
+ if dag_id != "~":
+ query = query.filter(TI.dag_id == dag_id)
+ if dag_run_id != "~":
+ query = query.join(DR, and_(TI.dag_id == DR.dag_id, TI.execution_date == DR.execution_date))
+ query = query.filter(DR.run_id == dag_run_id)
+ query = _apply_range_filter(
+ query,
+ key=DR.execution_date,
+ value_range=(execution_date_gte, execution_date_lte),
+ )
+ query = _apply_range_filter(query, key=TI.start_date, value_range=(start_date_gte, start_date_lte))
+ query = _apply_range_filter(query, key=TI.end_date, value_range=(end_date_gte, end_date_lte))
+ query = _apply_range_filter(query, key=TI.duration, value_range=(duration_gte, duration_lte))
+ query = _apply_array_filter(query, key=TI.state, values=state)
+ query = _apply_array_filter(query, key=TI.pool, values=pool)
+ query = _apply_array_filter(query, key=TI.queue, values=queue)
+
+ query = query.join(
+ SlaMiss,
+ and_(
+ SlaMiss.dag_id == TI.dag_id,
+ SlaMiss.task_id == TI.task_id,
+ SlaMiss.execution_date == TI.execution_date,
+ ),
+ isouter=True,
+ ).add_entity(SlaMiss)
+ task_instances = query.offset(offset).limit(limit).all()
+ total_entries = session.query(
+ func.count(func.distinct(TI.dag_id, TI.execution_date, TI.task_id))
+ ).scalar()
Review comment:
It seems to me that the count query doesn't need join, so we can simplify everything and improve SQL compatibility.
----------------------------------------------------------------
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 #9597: [WIP] Add read-only endpoints for task instances
Posted by GitBox <gi...@apache.org>.
mik-laj commented on pull request #9597:
URL: https://github.com/apache/airflow/pull/9597#issuecomment-704136784
I have a problem with this solution. I added one test that tests the SLA endpoint on this branch. Can you take a look at 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] mik-laj commented on a change in pull request #9597: [WIP] Add read-only endpoints for task instances
Posted by GitBox <gi...@apache.org>.
mik-laj commented on a change in pull request #9597:
URL: https://github.com/apache/airflow/pull/9597#discussion_r448587313
##########
File path: airflow/api_connexion/endpoints/task_instance_endpoint.py
##########
@@ -14,23 +14,110 @@
# KIND, either express or implied. See the License for the
# specific language governing permissions and limitations
# under the License.
+from typing import Any, List, Optional, Tuple
-# TODO(mik-laj): We have to implement it.
-# Do you want to help? Please look at: https://github.com/apache/airflow/issues/8132
+from sqlalchemy import and_, func, or_
+from airflow.api_connexion.exceptions import NotFound
+from airflow.api_connexion.parameters import format_datetime, format_parameters
+from airflow.api_connexion.schemas.task_instance_schema import (
+ TaskInstanceCollection, task_instance_collection_schema, task_instance_schema,
+)
+from airflow.models.dagrun import DagRun as DR
+from airflow.models.taskinstance import TaskInstance as TI
+from airflow.utils.session import provide_session
+
+
+@provide_session
+def get_task_instance(dag_id: str, dag_run_id: str, task_id: str, session=None):
+ query = (
+ session.query(TI)
+ .filter(TI.dag_id == dag_id)
+ .join(DR, and_(TI.dag_id == DR.dag_id, TI.execution_date == DR.execution_date))
+ .filter(DR.run_id == dag_run_id)
+ .filter(TI.task_id == task_id)
+ )
+
+ task_instance = query.one_or_none()
+
+ if task_instance is None:
+ raise NotFound("Task instance not found")
+
+ return task_instance_schema.dump(task_instance)
-def get_task_instance():
- """
- Get a task instance
- """
- raise NotImplementedError("Not implemented yet.")
+def _apply_array_filter(query, key, values):
Review comment:
I do not know if I understand correctly. Can you say more?
----------------------------------------------------------------
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 #9597: [WIP] Add read-only endpoints for task instances
Posted by GitBox <gi...@apache.org>.
OmairK commented on a change in pull request #9597:
URL: https://github.com/apache/airflow/pull/9597#discussion_r501395345
##########
File path: airflow/api_connexion/endpoints/task_instance_endpoint.py
##########
@@ -14,39 +14,243 @@
# KIND, either express or implied. See the License for the
# specific language governing permissions and limitations
# under the License.
+from typing import Any, List, Optional, Tuple
+
+from flask import request
+from marshmallow import ValidationError
+from sqlalchemy import and_, func, or_
-# TODO(mik-laj): We have to implement it.
-# Do you want to help? Please look at: https://github.com/apache/airflow/issues/8132
from airflow.api_connexion import security
+from airflow.api_connexion.exceptions import NotFound, BadRequest
+from airflow.api_connexion.parameters import format_datetime, format_parameters
+from airflow.api_connexion.schemas.task_instance_schema import (
+ clear_task_instance_form,
+ TaskInstanceCollection,
+ task_instance_collection_schema,
+ task_instance_schema,
+ task_instance_batch_form,
+ task_instance_reference_collection_schema,
+ TaskInstanceReferenceCollection,
+)
+from airflow.configuration import conf
+from airflow.models.dagrun import DagRun as DR
+from airflow.models.taskinstance import clear_task_instances, TaskInstance as TI
+from airflow.models.dag import DagModel
+from airflow.models.dagbag import DagBag
+from airflow.models import SlaMiss
+from airflow.utils.state import State
+from airflow.utils.session import provide_session
@security.requires_authentication
-def get_task_instance():
- """
- Get a task instance
- """
- raise NotImplementedError("Not implemented yet.")
+@provide_session
+def get_task_instance(dag_id: str, dag_run_id: str, task_id: str, session=None):
+ """Get task instance"""
+ query = (
+ session.query(TI)
+ .filter(TI.dag_id == dag_id)
+ .join(DR, and_(TI.dag_id == DR.dag_id, TI.execution_date == DR.execution_date))
+ .filter(DR.run_id == dag_run_id)
+ .filter(TI.task_id == task_id)
+ .join(
+ SlaMiss,
+ and_(
+ SlaMiss.dag_id == TI.dag_id,
+ SlaMiss.execution_date == TI.execution_date,
+ SlaMiss.task_id == TI.task_id,
+ ),
+ isouter=True,
+ )
+ .add_entity(SlaMiss)
+ )
+ task_instance = query.one_or_none()
+ if task_instance is None:
+ raise NotFound("Task instance not found")
+
+ return task_instance_schema.dump(task_instance)
+
+
+def _apply_array_filter(query, key, values):
+ if values is not None:
+ query = query.filter(or_(*[key == v for v in values]))
+ return query
+
+
+def _apply_range_filter(query, key, value_range: Tuple[Any, Any]):
+ gte_value, lte_value = value_range
+ if gte_value is not None:
+ query = query.filter(key >= gte_value)
+ if lte_value is not None:
+ query = query.filter(key <= lte_value)
+ return query
+@format_parameters(
+ {
+ "execution_date_gte": format_datetime,
+ "execution_date_lte": format_datetime,
+ "start_date_gte": format_datetime,
+ "start_date_lte": format_datetime,
+ "end_date_gte": format_datetime,
+ "end_date_lte": format_datetime,
+ }
+)
@security.requires_authentication
-def get_task_instances():
+@provide_session
+def get_task_instances(
+ limit: int,
+ dag_id: Optional[str] = None,
+ dag_run_id: Optional[str] = None,
+ execution_date_gte: Optional[str] = None,
+ execution_date_lte: Optional[str] = None,
+ start_date_gte: Optional[str] = None,
+ start_date_lte: Optional[str] = None,
+ end_date_gte: Optional[str] = None,
+ end_date_lte: Optional[str] = None,
+ duration_gte: Optional[float] = None,
+ duration_lte: Optional[float] = None,
+ state: Optional[str] = None,
+ pool: Optional[List[str]] = None,
+ queue: Optional[List[str]] = None,
+ offset: Optional[int] = None,
+ session=None,
+): # pylint: disable=too-many-arguments
"""
- Get list of task instances of DAG.
+ Get list of task instances.
"""
- raise NotImplementedError("Not implemented yet.")
+ query = session.query(TI)
+
+ if dag_id != "~":
+ query = query.filter(TI.dag_id == dag_id)
+ if dag_run_id != "~":
+ query = query.join(DR, and_(TI.dag_id == DR.dag_id, TI.execution_date == DR.execution_date))
+ query = query.filter(DR.run_id == dag_run_id)
+ query = _apply_range_filter(
+ query,
+ key=DR.execution_date,
+ value_range=(execution_date_gte, execution_date_lte),
+ )
+ query = _apply_range_filter(query, key=TI.start_date, value_range=(start_date_gte, start_date_lte))
+ query = _apply_range_filter(query, key=TI.end_date, value_range=(end_date_gte, end_date_lte))
+ query = _apply_range_filter(query, key=TI.duration, value_range=(duration_gte, duration_lte))
+ query = _apply_array_filter(query, key=TI.state, values=state)
+ query = _apply_array_filter(query, key=TI.pool, values=pool)
+ query = _apply_array_filter(query, key=TI.queue, values=queue)
+
+ query = query.join(
+ SlaMiss,
+ and_(
+ SlaMiss.dag_id == TI.dag_id,
+ SlaMiss.task_id == TI.task_id,
+ SlaMiss.execution_date == TI.execution_date,
+ ),
+ isouter=True,
+ ).add_entity(SlaMiss)
+ task_instances = query.offset(offset).limit(limit).all()
+ total_entries = session.query(
+ func.count(func.distinct(TI.dag_id, TI.execution_date, TI.task_id))
+ ).scalar()
Review comment:
@mik-laj Passing multiple columns to distinct works perfectly with postgres but raises error in case of sqlite how should we handle this case.
----------------------------------------------------------------
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 #9597: [WIP] Add read-only endpoints for task instances
Posted by GitBox <gi...@apache.org>.
OmairK commented on a change in pull request #9597:
URL: https://github.com/apache/airflow/pull/9597#discussion_r478796025
##########
File path: airflow/api_connexion/schemas/task_instance_schema.py
##########
@@ -0,0 +1,147 @@
+# 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, ValidationError, post_load
+from marshmallow_sqlalchemy import SQLAlchemySchema, auto_field
+from sqlalchemy import and_
+
+from airflow.api_connexion.schemas.enum_schemas import TaskInstanceStateField
+from airflow.api_connexion.schemas.sla_miss_schema import SlaMissSchema
+from airflow.models import DagRun, TaskInstance, SlaMiss
+from airflow.utils.session import create_session
+
+
+class TaskInstanceSchema(Schema):
+ """Task instance schema"""
+
+ task_id = fields.Str()
+ dag_id = fields.Str()
+ execution_date = fields.DateTime()
+ start_date = fields.DateTime()
+ end_date = fields.DateTime()
+ duration = fields.Float()
+ state = TaskInstanceStateField()
+ _try_number = fields.Int(data_key="try_number")
+ max_tries = fields.Int()
+ hostname = fields.Str()
+ unixname = fields.Str()
+ pool = fields.Str()
+ pool_slots = fields.Int()
+ queue = fields.Str()
+ priority_weight = fields.Int()
+ operator = fields.Str()
+ queued_dttm = fields.DateTime(data_key="queued_when")
+ pid = fields.Int()
+ executor_config = fields.Str()
+ sla_miss = fields.Method("get_sla_miss")
+
+ @staticmethod
+ def get_sla_miss(obj: TaskInstance):
+ with create_session() as session:
+ sla_miss = session.query(SlaMiss).filter(
Review comment:
Hey @mik-laj we are not using relationships in the model so I was planning to use explicit joins to overcome this n + 1 problem the only downside of that it is returns a tuple of object `(<task instance>,<sla miss>)` so I was planning to overwrite [`get_attribute`](https://marshmallow.readthedocs.io/en/stable/api_reference.html#marshmallow.Schema.get_attribute). Let me know if this sounds like a good idea. I will be doing something similar here
```
class TaskInstanceReferenceSchema(Schema):
"""Schema for the task instance reference schema"""
task_id = fields.Str()
dag_run_id = fields.Method('get_run_id')
dag_id = fields.Str()
execution_date = fields.DateTime()
@staticmethod
def get_run_id(obj: TaskInstance):
with create_session() as session:
return session.query(DagRun).filter(DagRun.dag_id==obj.dag_id, DagRun.execution_date==obj.execution_date).one_or_none().run_id
```
----------------------------------------------------------------
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 #9597: Add endpoints for task instances
Posted by GitBox <gi...@apache.org>.
kaxil commented on a change in pull request #9597:
URL: https://github.com/apache/airflow/pull/9597#discussion_r503504379
##########
File path: airflow/api_connexion/endpoints/task_instance_endpoint.py
##########
@@ -14,46 +14,260 @@
# KIND, either express or implied. See the License for the
# specific language governing permissions and limitations
# under the License.
+from typing import Any, List, Optional, Tuple
+
+from flask import request
+from marshmallow import ValidationError
+from sqlalchemy import and_, func
-# TODO(mik-laj): We have to implement it.
-# Do you want to help? Please look at: https://github.com/apache/airflow/issues/8132
from airflow.api_connexion import security
+from airflow.api_connexion.exceptions import NotFound, BadRequest
+from airflow.api_connexion.parameters import format_datetime, format_parameters
+from airflow.api_connexion.schemas.task_instance_schema import (
+ clear_task_instance_form,
+ TaskInstanceCollection,
+ task_instance_collection_schema,
+ task_instance_schema,
+ task_instance_batch_form,
+ task_instance_reference_collection_schema,
+ TaskInstanceReferenceCollection,
+)
+from airflow.configuration import conf
+from airflow.models.dagrun import DagRun as DR
+from airflow.models.taskinstance import clear_task_instances, TaskInstance as TI
+from airflow.models.dag import DagModel
+from airflow.models.dagbag import DagBag
+from airflow.models import SlaMiss
+from airflow.utils.state import State
+from airflow.utils.session import provide_session
-@security.requires_access([("can_read", "Dag"), ("can_read", "DagRun"), ("can_read", "Task")])
-def get_task_instance():
- """
- Get a task instance
- """
- raise NotImplementedError("Not implemented yet.")
+@security.requires_access(
+ [
+ ("can_read", "Dag"),
+ ("can_read", "DagRun"),
+ ('can_read', 'Task'),
+ ]
+)
+@provide_session
+def get_task_instance(dag_id: str, dag_run_id: str, task_id: str, session=None):
+ """Get task instance"""
+ query = (
+ session.query(TI)
+ .filter(TI.dag_id == dag_id)
+ .join(DR, and_(TI.dag_id == DR.dag_id, TI.execution_date == DR.execution_date))
+ .filter(DR.run_id == dag_run_id)
+ .filter(TI.task_id == task_id)
+ .join(
+ SlaMiss,
+ and_(
+ SlaMiss.dag_id == TI.dag_id,
+ SlaMiss.execution_date == TI.execution_date,
+ SlaMiss.task_id == TI.task_id,
+ ),
+ isouter=True,
+ )
+ .add_entity(SlaMiss)
+ )
+ task_instance = query.one_or_none()
+ if task_instance is None:
+ raise NotFound("Task instance not found")
+ return task_instance_schema.dump(task_instance)
-@security.requires_access([("can_read", "Dag"), ("can_read", "DagRun"), ("can_read", "Task")])
-def get_task_instances():
- """
- Get list of task instances of DAG.
- """
- raise NotImplementedError("Not implemented yet.")
+
+def _apply_array_filter(query, key, values):
+ if values is not None:
+ query = query.filter(key.in_(values))
+ return query
+
+
+def _apply_range_filter(query, key, value_range: Tuple[Any, Any]):
+ gte_value, lte_value = value_range
+ if gte_value is not None:
+ query = query.filter(key >= gte_value)
+ if lte_value is not None:
+ query = query.filter(key <= lte_value)
+ return query
+@format_parameters(
+ {
+ "execution_date_gte": format_datetime,
+ "execution_date_lte": format_datetime,
+ "start_date_gte": format_datetime,
+ "start_date_lte": format_datetime,
+ "end_date_gte": format_datetime,
+ "end_date_lte": format_datetime,
+ }
+)
@security.requires_access(
[
- ('can_read', 'Dag'),
- ('can_read', 'DagRun'),
+ ("can_read", "Dag"),
+ ("can_read", "DagRun"),
('can_read', 'Task'),
Review comment:
Can we change single to double quotes here too for consistency
----------------------------------------------------------------
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 #9597: [WIP] Add read-only endpoints for task instances
Posted by GitBox <gi...@apache.org>.
mik-laj edited a comment on pull request #9597:
URL: https://github.com/apache/airflow/pull/9597#issuecomment-704136784
I have a problem with this solution. I added one test that tests this endpoint with task instance and SLAMiss on this branch. Can you take a look at 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] mik-laj commented on a change in pull request #9597: [WIP] Add read-only endpoints for task instances
Posted by GitBox <gi...@apache.org>.
mik-laj commented on a change in pull request #9597:
URL: https://github.com/apache/airflow/pull/9597#discussion_r478797187
##########
File path: airflow/api_connexion/schemas/task_instance_schema.py
##########
@@ -0,0 +1,147 @@
+# 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, ValidationError, post_load
+from marshmallow_sqlalchemy import SQLAlchemySchema, auto_field
+from sqlalchemy import and_
+
+from airflow.api_connexion.schemas.enum_schemas import TaskInstanceStateField
+from airflow.api_connexion.schemas.sla_miss_schema import SlaMissSchema
+from airflow.models import DagRun, TaskInstance, SlaMiss
+from airflow.utils.session import create_session
+
+
+class TaskInstanceSchema(Schema):
+ """Task instance schema"""
+
+ task_id = fields.Str()
+ dag_id = fields.Str()
+ execution_date = fields.DateTime()
+ start_date = fields.DateTime()
+ end_date = fields.DateTime()
+ duration = fields.Float()
+ state = TaskInstanceStateField()
+ _try_number = fields.Int(data_key="try_number")
+ max_tries = fields.Int()
+ hostname = fields.Str()
+ unixname = fields.Str()
+ pool = fields.Str()
+ pool_slots = fields.Int()
+ queue = fields.Str()
+ priority_weight = fields.Int()
+ operator = fields.Str()
+ queued_dttm = fields.DateTime(data_key="queued_when")
+ pid = fields.Int()
+ executor_config = fields.Str()
+ sla_miss = fields.Method("get_sla_miss")
+
+ @staticmethod
+ def get_sla_miss(obj: TaskInstance):
+ with create_session() as session:
+ sla_miss = session.query(SlaMiss).filter(
Review comment:
if you need it, we can start using relationships.
----------------------------------------------------------------
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 #9597: Add endpoints for task instances
Posted by GitBox <gi...@apache.org>.
mik-laj commented on a change in pull request #9597:
URL: https://github.com/apache/airflow/pull/9597#discussion_r503755505
##########
File path: airflow/api_connexion/endpoints/task_instance_endpoint.py
##########
@@ -14,46 +14,260 @@
# KIND, either express or implied. See the License for the
# specific language governing permissions and limitations
# under the License.
+from typing import Any, List, Optional, Tuple
+
+from flask import request
+from marshmallow import ValidationError
+from sqlalchemy import and_, func
-# TODO(mik-laj): We have to implement it.
-# Do you want to help? Please look at: https://github.com/apache/airflow/issues/8132
from airflow.api_connexion import security
+from airflow.api_connexion.exceptions import NotFound, BadRequest
+from airflow.api_connexion.parameters import format_datetime, format_parameters
+from airflow.api_connexion.schemas.task_instance_schema import (
+ clear_task_instance_form,
+ TaskInstanceCollection,
+ task_instance_collection_schema,
+ task_instance_schema,
+ task_instance_batch_form,
+ task_instance_reference_collection_schema,
+ TaskInstanceReferenceCollection,
+)
+from airflow.configuration import conf
+from airflow.models.dagrun import DagRun as DR
+from airflow.models.taskinstance import clear_task_instances, TaskInstance as TI
+from airflow.models.dag import DagModel
+from airflow.models.dagbag import DagBag
+from airflow.models import SlaMiss
+from airflow.utils.state import State
+from airflow.utils.session import provide_session
-@security.requires_access([("can_read", "Dag"), ("can_read", "DagRun"), ("can_read", "Task")])
-def get_task_instance():
- """
- Get a task instance
- """
- raise NotImplementedError("Not implemented yet.")
+@security.requires_access(
+ [
+ ("can_read", "Dag"),
+ ("can_read", "DagRun"),
+ ('can_read', 'Task'),
+ ]
+)
+@provide_session
+def get_task_instance(dag_id: str, dag_run_id: str, task_id: str, session=None):
+ """Get task instance"""
+ query = (
+ session.query(TI)
+ .filter(TI.dag_id == dag_id)
+ .join(DR, and_(TI.dag_id == DR.dag_id, TI.execution_date == DR.execution_date))
+ .filter(DR.run_id == dag_run_id)
+ .filter(TI.task_id == task_id)
+ .join(
+ SlaMiss,
+ and_(
+ SlaMiss.dag_id == TI.dag_id,
+ SlaMiss.execution_date == TI.execution_date,
+ SlaMiss.task_id == TI.task_id,
+ ),
+ isouter=True,
+ )
+ .add_entity(SlaMiss)
+ )
+ task_instance = query.one_or_none()
+ if task_instance is None:
+ raise NotFound("Task instance not found")
+ return task_instance_schema.dump(task_instance)
-@security.requires_access([("can_read", "Dag"), ("can_read", "DagRun"), ("can_read", "Task")])
-def get_task_instances():
- """
- Get list of task instances of DAG.
- """
- raise NotImplementedError("Not implemented yet.")
+
+def _apply_array_filter(query, key, values):
+ if values is not None:
+ query = query.filter(key.in_(values))
+ return query
+
+
+def _apply_range_filter(query, key, value_range: Tuple[Any, Any]):
+ gte_value, lte_value = value_range
+ if gte_value is not None:
+ query = query.filter(key >= gte_value)
+ if lte_value is not None:
+ query = query.filter(key <= lte_value)
+ return query
+@format_parameters(
+ {
+ "execution_date_gte": format_datetime,
+ "execution_date_lte": format_datetime,
+ "start_date_gte": format_datetime,
+ "start_date_lte": format_datetime,
+ "end_date_gte": format_datetime,
+ "end_date_lte": format_datetime,
+ }
+)
@security.requires_access(
[
- ('can_read', 'Dag'),
- ('can_read', 'DagRun'),
+ ("can_read", "Dag"),
+ ("can_read", "DagRun"),
('can_read', 'Task'),
Review comment:
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] github-actions[bot] commented on pull request #9597: Add endpoints for task instances
Posted by GitBox <gi...@apache.org>.
github-actions[bot] commented on pull request #9597:
URL: https://github.com/apache/airflow/pull/9597#issuecomment-706169947
[The Workflow run](https://github.com/apache/airflow/actions/runs/297511179) is cancelling this PR. It has some failed jobs matching ^Pylint$,^Static checks$,^Build docs$,^Spell check docs$,^Backport packages$,^Checks: Helm tests$,^Test OpenAPI*.
----------------------------------------------------------------
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 #9597: Add endpoints for task instances
Posted by GitBox <gi...@apache.org>.
mik-laj commented on a change in pull request #9597:
URL: https://github.com/apache/airflow/pull/9597#discussion_r502515545
##########
File path: airflow/api_connexion/endpoints/task_instance_endpoint.py
##########
@@ -14,39 +14,234 @@
# KIND, either express or implied. See the License for the
# specific language governing permissions and limitations
# under the License.
+from typing import Any, List, Optional, Tuple
+
+from flask import request
+from marshmallow import ValidationError
+from sqlalchemy import and_, func, or_
-# TODO(mik-laj): We have to implement it.
-# Do you want to help? Please look at: https://github.com/apache/airflow/issues/8132
from airflow.api_connexion import security
+from airflow.api_connexion.exceptions import NotFound, BadRequest
+from airflow.api_connexion.parameters import format_datetime, format_parameters
+from airflow.api_connexion.schemas.task_instance_schema import (
+ clear_task_instance_form,
+ TaskInstanceCollection,
+ task_instance_collection_schema,
+ task_instance_schema,
+ task_instance_batch_form,
+ task_instance_reference_collection_schema,
+ TaskInstanceReferenceCollection,
+)
+from airflow.configuration import conf
+from airflow.models.dagrun import DagRun as DR
+from airflow.models.taskinstance import clear_task_instances, TaskInstance as TI
+from airflow.models.dag import DagModel
+from airflow.models.dagbag import DagBag
+from airflow.models import SlaMiss
+from airflow.utils.state import State
+from airflow.utils.session import provide_session
@security.requires_authentication
-def get_task_instance():
- """
- Get a task instance
- """
- raise NotImplementedError("Not implemented yet.")
+@provide_session
+def get_task_instance(dag_id: str, dag_run_id: str, task_id: str, session=None):
+ """Get task instance"""
+ query = (
+ session.query(TI)
+ .filter(TI.dag_id == dag_id)
+ .join(DR, and_(TI.dag_id == DR.dag_id, TI.execution_date == DR.execution_date))
+ .filter(DR.run_id == dag_run_id)
+ .filter(TI.task_id == task_id)
+ .join(
+ SlaMiss,
+ and_(
+ SlaMiss.dag_id == TI.dag_id,
+ SlaMiss.execution_date == TI.execution_date,
+ SlaMiss.task_id == TI.task_id,
+ ),
+ isouter=True,
+ )
+ .add_entity(SlaMiss)
+ )
+ task_instance = query.one_or_none()
+
+ if task_instance is None:
+ raise NotFound("Task instance not found")
+
+ return task_instance_schema.dump(task_instance)
+
+
+def _apply_array_filter(query, key, values):
+ if values is not None:
+ query = query.filter(or_(*[key == v for v in values]))
+ return query
+
+
+def _apply_range_filter(query, key, value_range: Tuple[Any, Any]):
+ gte_value, lte_value = value_range
+ if gte_value is not None:
+ query = query.filter(key >= gte_value)
+ if lte_value is not None:
+ query = query.filter(key <= lte_value)
+ return query
+@format_parameters(
+ {
+ "execution_date_gte": format_datetime,
+ "execution_date_lte": format_datetime,
+ "start_date_gte": format_datetime,
+ "start_date_lte": format_datetime,
+ "end_date_gte": format_datetime,
+ "end_date_lte": format_datetime,
+ }
+)
@security.requires_authentication
-def get_task_instances():
+@provide_session
+def get_task_instances(
+ limit: int,
+ dag_id: Optional[str] = None,
+ dag_run_id: Optional[str] = None,
+ execution_date_gte: Optional[str] = None,
+ execution_date_lte: Optional[str] = None,
+ start_date_gte: Optional[str] = None,
+ start_date_lte: Optional[str] = None,
+ end_date_gte: Optional[str] = None,
+ end_date_lte: Optional[str] = None,
+ duration_gte: Optional[float] = None,
+ duration_lte: Optional[float] = None,
+ state: Optional[str] = None,
+ pool: Optional[List[str]] = None,
+ queue: Optional[List[str]] = None,
+ offset: Optional[int] = None,
+ session=None,
+): # pylint: disable=too-many-arguments
"""
- Get list of task instances of DAG.
+ Get list of task instances.
"""
- raise NotImplementedError("Not implemented yet.")
+ query = session.query(TI)
+
+ if dag_id != "~":
+ query = query.filter(TI.dag_id == dag_id)
+ if dag_run_id != "~":
+ query = query.join(DR, and_(TI.dag_id == DR.dag_id, TI.execution_date == DR.execution_date))
+ query = query.filter(DR.run_id == dag_run_id)
+ query = _apply_range_filter(
+ query, key=DR.execution_date, value_range=(execution_date_gte, execution_date_lte),
+ )
+ query = _apply_range_filter(query, key=TI.start_date, value_range=(start_date_gte, start_date_lte))
+ query = _apply_range_filter(query, key=TI.end_date, value_range=(end_date_gte, end_date_lte))
+ query = _apply_range_filter(query, key=TI.duration, value_range=(duration_gte, duration_lte))
+ query = _apply_array_filter(query, key=TI.state, values=state)
+ query = _apply_array_filter(query, key=TI.pool, values=pool)
+ query = _apply_array_filter(query, key=TI.queue, values=queue)
+
+ query = query.join(
+ SlaMiss,
+ and_(
+ SlaMiss.dag_id == TI.dag_id,
+ SlaMiss.task_id == TI.task_id,
+ SlaMiss.execution_date == TI.execution_date,
+ ),
+ isouter=True,
+ ).add_entity(SlaMiss)
+ task_instances = query.offset(offset).limit(limit).all()
+ total_entries = session.query(
+ func.count(func.distinct(TI.dag_id, TI.execution_date, TI.task_id))
+ ).scalar()
+ return task_instance_collection_schema.dump(
+ TaskInstanceCollection(task_instances=task_instances, total_entries=total_entries)
+ )
@security.requires_authentication
-def get_task_instances_batch():
+@provide_session
+def get_task_instances_batch(session=None):
"""
Get list of task instances.
"""
- raise NotImplementedError("Not implemented yet.")
+ body = request.get_json()
+ try:
+ data = task_instance_batch_form.load(body)
+ except ValidationError as err:
+ raise BadRequest(detail=str(err.messages))
+ query = session.query(TI)
+
+ query = _apply_array_filter(query, key=TI.dag_id, values=data["dag_ids"])
+ query = _apply_range_filter(
+ query, key=TI.execution_date, value_range=(data["execution_date_gte"], data["execution_date_lte"]),
+ )
+ query = _apply_range_filter(
+ query, key=TI.start_date, value_range=(data["start_date_gte"], data["start_date_lte"]),
+ )
+ query = _apply_range_filter(
+ query, key=TI.end_date, value_range=(data["end_date_gte"], data["end_date_lte"])
+ )
+ query = _apply_range_filter(
+ query, key=TI.duration, value_range=(data["duration_gte"], data["duration_lte"])
+ )
+ query = _apply_array_filter(query, key=TI.state, values=data["state"])
+ query = _apply_array_filter(query, key=TI.pool, values=data["pool"])
+ query = _apply_array_filter(query, key=TI.queue, values=data["queue"])
+
+ query = query.join(
+ SlaMiss,
+ and_(
+ SlaMiss.dag_id == TI.dag_id,
+ SlaMiss.task_id == TI.task_id,
+ SlaMiss.execution_date == TI.execution_date,
+ ),
+ isouter=True,
+ ).add_entity(SlaMiss)
+ task_instances = query.all()
+ total_entries = session.query(
+ func.count(func.distinct(TI.dag_id, TI.execution_date, TI.task_id))
+ ).scalar()
+
+ return task_instance_collection_schema.dump(
+ TaskInstanceCollection(task_instances=task_instances, total_entries=total_entries)
+ )
@security.requires_authentication
-def post_clear_task_instances():
+@provide_session
+def post_clear_task_instances(dag_id: str, session=None):
Review comment:
Unfortunately not. I will add them in a moment.
----------------------------------------------------------------
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] jhtimmins commented on a change in pull request #9597: [WIP] Add read-only endpoints for task instances
Posted by GitBox <gi...@apache.org>.
jhtimmins commented on a change in pull request #9597:
URL: https://github.com/apache/airflow/pull/9597#discussion_r448127145
##########
File path: airflow/api_connexion/endpoints/task_instance_endpoint.py
##########
@@ -14,23 +14,110 @@
# KIND, either express or implied. See the License for the
# specific language governing permissions and limitations
# under the License.
+from typing import Any, List, Optional, Tuple
-# TODO(mik-laj): We have to implement it.
-# Do you want to help? Please look at: https://github.com/apache/airflow/issues/8132
+from sqlalchemy import and_, func, or_
+from airflow.api_connexion.exceptions import NotFound
+from airflow.api_connexion.parameters import format_datetime, format_parameters
+from airflow.api_connexion.schemas.task_instance_schema import (
+ TaskInstanceCollection, task_instance_collection_schema, task_instance_schema,
+)
+from airflow.models.dagrun import DagRun as DR
+from airflow.models.taskinstance import TaskInstance as TI
+from airflow.utils.session import provide_session
+
+
+@provide_session
+def get_task_instance(dag_id: str, dag_run_id: str, task_id: str, session=None):
+ query = (
+ session.query(TI)
+ .filter(TI.dag_id == dag_id)
+ .join(DR, and_(TI.dag_id == DR.dag_id, TI.execution_date == DR.execution_date))
+ .filter(DR.run_id == dag_run_id)
+ .filter(TI.task_id == task_id)
+ )
+
+ task_instance = query.one_or_none()
+
+ if task_instance is None:
+ raise NotFound("Task instance not found")
Review comment:
I could be missing it, but are we catching this Exception anywhere?
##########
File path: airflow/api_connexion/endpoints/task_instance_endpoint.py
##########
@@ -14,23 +14,110 @@
# KIND, either express or implied. See the License for the
# specific language governing permissions and limitations
# under the License.
+from typing import Any, List, Optional, Tuple
-# TODO(mik-laj): We have to implement it.
-# Do you want to help? Please look at: https://github.com/apache/airflow/issues/8132
+from sqlalchemy import and_, func, or_
+from airflow.api_connexion.exceptions import NotFound
+from airflow.api_connexion.parameters import format_datetime, format_parameters
+from airflow.api_connexion.schemas.task_instance_schema import (
+ TaskInstanceCollection, task_instance_collection_schema, task_instance_schema,
+)
+from airflow.models.dagrun import DagRun as DR
+from airflow.models.taskinstance import TaskInstance as TI
+from airflow.utils.session import provide_session
+
+
+@provide_session
+def get_task_instance(dag_id: str, dag_run_id: str, task_id: str, session=None):
+ query = (
+ session.query(TI)
+ .filter(TI.dag_id == dag_id)
+ .join(DR, and_(TI.dag_id == DR.dag_id, TI.execution_date == DR.execution_date))
+ .filter(DR.run_id == dag_run_id)
+ .filter(TI.task_id == task_id)
+ )
+
+ task_instance = query.one_or_none()
+
+ if task_instance is None:
+ raise NotFound("Task instance not found")
+
+ return task_instance_schema.dump(task_instance)
-def get_task_instance():
- """
- Get a task instance
- """
- raise NotImplementedError("Not implemented yet.")
+def _apply_array_filter(query, key, values):
Review comment:
This filter seems somewhat ambiguous. I think the name could better express the purpose.
##########
File path: airflow/api_connexion/endpoints/task_instance_endpoint.py
##########
@@ -14,23 +14,110 @@
# KIND, either express or implied. See the License for the
# specific language governing permissions and limitations
# under the License.
+from typing import Any, List, Optional, Tuple
-# TODO(mik-laj): We have to implement it.
-# Do you want to help? Please look at: https://github.com/apache/airflow/issues/8132
+from sqlalchemy import and_, func, or_
+from airflow.api_connexion.exceptions import NotFound
+from airflow.api_connexion.parameters import format_datetime, format_parameters
+from airflow.api_connexion.schemas.task_instance_schema import (
+ TaskInstanceCollection, task_instance_collection_schema, task_instance_schema,
+)
+from airflow.models.dagrun import DagRun as DR
+from airflow.models.taskinstance import TaskInstance as TI
+from airflow.utils.session import provide_session
+
+
+@provide_session
+def get_task_instance(dag_id: str, dag_run_id: str, task_id: str, session=None):
+ query = (
+ session.query(TI)
+ .filter(TI.dag_id == dag_id)
+ .join(DR, and_(TI.dag_id == DR.dag_id, TI.execution_date == DR.execution_date))
+ .filter(DR.run_id == dag_run_id)
+ .filter(TI.task_id == task_id)
+ )
+
+ task_instance = query.one_or_none()
+
+ if task_instance is None:
+ raise NotFound("Task instance not found")
+
+ return task_instance_schema.dump(task_instance)
-def get_task_instance():
- """
- Get a task instance
- """
- raise NotImplementedError("Not implemented yet.")
+def _apply_array_filter(query, key, values):
+ if values is not None:
+ query = query.filter(or_(*[key == v for v in values]))
+ return query
-def get_task_instances():
+
+def _apply_range_filter(query, key, value_range: Tuple[Any, Any]):
+ gte_value, lte_value = value_range
+ if gte_value is not None:
+ query = query.filter(key >= gte_value)
+ if lte_value is not None:
+ query = query.filter(key <= lte_value)
+ return query
+
+
+@format_parameters(
+ {
+ 'start_date_gte': format_datetime,
+ 'start_date_lte': format_datetime,
+ 'execution_date_gte': format_datetime,
+ 'execution_date_lte': format_datetime,
+ 'end_date_gte': format_datetime,
+ 'end_date_lte': format_datetime,
+ }
+)
+@provide_session
+def get_task_instances(
+ limit: int,
+ dag_id: Optional[str] = None,
+ dag_run_id: Optional[str] = None,
+ execution_date_gte: Optional[str] = None,
+ execution_date_lte: Optional[str] = None,
+ start_date_gte: Optional[str] = None,
+ start_date_lte: Optional[str] = None,
+ end_date_gte: Optional[str] = None,
+ end_date_lte: Optional[str] = None,
+ duration_gte: Optional[float] = None,
+ duration_lte: Optional[float] = None,
+ state: Optional[str] = None,
+ pool: Optional[List[str]] = None,
+ queue: Optional[List[str]] = None,
+ offset: Optional[int] = None,
+ session=None,
+):
"""
- Get list of task instances of DAG.
+ Get list of a task instances
Review comment:
```suggestion
Get list of task instances
```
##########
File path: airflow/api_connexion/endpoints/task_instance_endpoint.py
##########
@@ -14,23 +14,110 @@
# KIND, either express or implied. See the License for the
# specific language governing permissions and limitations
# under the License.
+from typing import Any, List, Optional, Tuple
-# TODO(mik-laj): We have to implement it.
-# Do you want to help? Please look at: https://github.com/apache/airflow/issues/8132
+from sqlalchemy import and_, func, or_
+from airflow.api_connexion.exceptions import NotFound
+from airflow.api_connexion.parameters import format_datetime, format_parameters
+from airflow.api_connexion.schemas.task_instance_schema import (
+ TaskInstanceCollection, task_instance_collection_schema, task_instance_schema,
+)
+from airflow.models.dagrun import DagRun as DR
+from airflow.models.taskinstance import TaskInstance as TI
+from airflow.utils.session import provide_session
+
+
+@provide_session
+def get_task_instance(dag_id: str, dag_run_id: str, task_id: str, session=None):
+ query = (
+ session.query(TI)
+ .filter(TI.dag_id == dag_id)
+ .join(DR, and_(TI.dag_id == DR.dag_id, TI.execution_date == DR.execution_date))
+ .filter(DR.run_id == dag_run_id)
+ .filter(TI.task_id == task_id)
+ )
+
+ task_instance = query.one_or_none()
+
+ if task_instance is None:
+ raise NotFound("Task instance not found")
+
+ return task_instance_schema.dump(task_instance)
-def get_task_instance():
- """
- Get a task instance
- """
- raise NotImplementedError("Not implemented yet.")
+def _apply_array_filter(query, key, values):
+ if values is not None:
+ query = query.filter(or_(*[key == v for v in values]))
+ return query
-def get_task_instances():
+
+def _apply_range_filter(query, key, value_range: Tuple[Any, Any]):
+ gte_value, lte_value = value_range
+ if gte_value is not None:
+ query = query.filter(key >= gte_value)
+ if lte_value is not None:
+ query = query.filter(key <= lte_value)
+ return query
+
+
+@format_parameters(
+ {
+ 'start_date_gte': format_datetime,
+ 'start_date_lte': format_datetime,
+ 'execution_date_gte': format_datetime,
+ 'execution_date_lte': format_datetime,
+ 'end_date_gte': format_datetime,
+ 'end_date_lte': format_datetime,
+ }
+)
+@provide_session
+def get_task_instances(
+ limit: int,
+ dag_id: Optional[str] = None,
+ dag_run_id: Optional[str] = None,
+ execution_date_gte: Optional[str] = None,
+ execution_date_lte: Optional[str] = None,
+ start_date_gte: Optional[str] = None,
+ start_date_lte: Optional[str] = None,
+ end_date_gte: Optional[str] = None,
+ end_date_lte: Optional[str] = None,
+ duration_gte: Optional[float] = None,
+ duration_lte: Optional[float] = None,
+ state: Optional[str] = None,
+ pool: Optional[List[str]] = None,
+ queue: Optional[List[str]] = None,
+ offset: Optional[int] = None,
+ session=None,
Review comment:
```suggestion
session = None,
```
##########
File path: airflow/api_connexion/endpoints/task_instance_endpoint.py
##########
@@ -14,23 +14,110 @@
# KIND, either express or implied. See the License for the
# specific language governing permissions and limitations
# under the License.
+from typing import Any, List, Optional, Tuple
-# TODO(mik-laj): We have to implement it.
-# Do you want to help? Please look at: https://github.com/apache/airflow/issues/8132
+from sqlalchemy import and_, func, or_
+from airflow.api_connexion.exceptions import NotFound
+from airflow.api_connexion.parameters import format_datetime, format_parameters
+from airflow.api_connexion.schemas.task_instance_schema import (
+ TaskInstanceCollection, task_instance_collection_schema, task_instance_schema,
+)
+from airflow.models.dagrun import DagRun as DR
+from airflow.models.taskinstance import TaskInstance as TI
+from airflow.utils.session import provide_session
+
+
+@provide_session
+def get_task_instance(dag_id: str, dag_run_id: str, task_id: str, session=None):
+ query = (
+ session.query(TI)
+ .filter(TI.dag_id == dag_id)
+ .join(DR, and_(TI.dag_id == DR.dag_id, TI.execution_date == DR.execution_date))
+ .filter(DR.run_id == dag_run_id)
+ .filter(TI.task_id == task_id)
+ )
+
+ task_instance = query.one_or_none()
+
+ if task_instance is None:
+ raise NotFound("Task instance not found")
+
+ return task_instance_schema.dump(task_instance)
-def get_task_instance():
- """
- Get a task instance
- """
- raise NotImplementedError("Not implemented yet.")
+def _apply_array_filter(query, key, values):
+ if values is not None:
+ query = query.filter(or_(*[key == v for v in values]))
+ return query
-def get_task_instances():
+
+def _apply_range_filter(query, key, value_range: Tuple[Any, Any]):
+ gte_value, lte_value = value_range
+ if gte_value is not None:
+ query = query.filter(key >= gte_value)
+ if lte_value is not None:
+ query = query.filter(key <= lte_value)
+ return query
+
+
+@format_parameters(
+ {
+ 'start_date_gte': format_datetime,
+ 'start_date_lte': format_datetime,
+ 'execution_date_gte': format_datetime,
+ 'execution_date_lte': format_datetime,
+ 'end_date_gte': format_datetime,
+ 'end_date_lte': format_datetime,
+ }
+)
+@provide_session
+def get_task_instances(
+ limit: int,
+ dag_id: Optional[str] = None,
+ dag_run_id: Optional[str] = None,
+ execution_date_gte: Optional[str] = None,
+ execution_date_lte: Optional[str] = None,
+ start_date_gte: Optional[str] = None,
+ start_date_lte: Optional[str] = None,
+ end_date_gte: Optional[str] = None,
+ end_date_lte: Optional[str] = None,
+ duration_gte: Optional[float] = None,
+ duration_lte: Optional[float] = None,
+ state: Optional[str] = None,
+ pool: Optional[List[str]] = None,
+ queue: Optional[List[str]] = None,
+ offset: Optional[int] = None,
+ session=None,
+):
"""
- Get list of task instances of DAG.
+ Get list of a task instances
"""
- raise NotImplementedError("Not implemented yet.")
+ query = session.query(TI)
+
+ if dag_id is not None:
+ query = query.filter(TI.dag_id == dag_id)
+ if dag_run_id is not None:
+ query = query.join(DR, and_(TI.dag_id == DR.dag_id, TI.execution_date == DR.execution_date))
+ query = query.filter(DR.run_id == dag_run_id)
+
+ query = _apply_range_filter(
+ query, key=DR.execution_date, value_range=(execution_date_gte, execution_date_lte)
+ )
+ query = _apply_range_filter(query, key=DR.start_date, value_range=(start_date_gte, start_date_lte))
+ query = _apply_range_filter(query, key=DR.end_date, value_range=(end_date_gte, end_date_lte))
+ query = _apply_range_filter(query, key=DR.end_date, value_range=(end_date_gte, end_date_lte))
Review comment:
Looks like the preceding two lines are duplicates
----------------------------------------------------------------
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 #9597: [WIP] Add read-only endpoints for task instances
Posted by GitBox <gi...@apache.org>.
OmairK commented on a change in pull request #9597:
URL: https://github.com/apache/airflow/pull/9597#discussion_r478799540
##########
File path: airflow/api_connexion/schemas/task_instance_schema.py
##########
@@ -0,0 +1,147 @@
+# 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, ValidationError, post_load
+from marshmallow_sqlalchemy import SQLAlchemySchema, auto_field
+from sqlalchemy import and_
+
+from airflow.api_connexion.schemas.enum_schemas import TaskInstanceStateField
+from airflow.api_connexion.schemas.sla_miss_schema import SlaMissSchema
+from airflow.models import DagRun, TaskInstance, SlaMiss
+from airflow.utils.session import create_session
+
+
+class TaskInstanceSchema(Schema):
+ """Task instance schema"""
+
+ task_id = fields.Str()
+ dag_id = fields.Str()
+ execution_date = fields.DateTime()
+ start_date = fields.DateTime()
+ end_date = fields.DateTime()
+ duration = fields.Float()
+ state = TaskInstanceStateField()
+ _try_number = fields.Int(data_key="try_number")
+ max_tries = fields.Int()
+ hostname = fields.Str()
+ unixname = fields.Str()
+ pool = fields.Str()
+ pool_slots = fields.Int()
+ queue = fields.Str()
+ priority_weight = fields.Int()
+ operator = fields.Str()
+ queued_dttm = fields.DateTime(data_key="queued_when")
+ pid = fields.Int()
+ executor_config = fields.Str()
+ sla_miss = fields.Method("get_sla_miss")
+
+ @staticmethod
+ def get_sla_miss(obj: TaskInstance):
+ with create_session() as session:
+ sla_miss = session.query(SlaMiss).filter(
Review comment:
Cool, in the meantime I will prepare this solution
----------------------------------------------------------------
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 #9597: [WIP] Add read-only endpoints for task instances
Posted by GitBox <gi...@apache.org>.
mik-laj commented on a change in pull request #9597:
URL: https://github.com/apache/airflow/pull/9597#discussion_r476043263
##########
File path: airflow/api_connexion/schemas/task_instance_schema.py
##########
@@ -0,0 +1,147 @@
+# 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, ValidationError, post_load
+from marshmallow_sqlalchemy import SQLAlchemySchema, auto_field
+from sqlalchemy import and_
+
+from airflow.api_connexion.schemas.enum_schemas import TaskInstanceStateField
+from airflow.api_connexion.schemas.sla_miss_schema import SlaMissSchema
+from airflow.models import DagRun, TaskInstance, SlaMiss
+from airflow.utils.session import create_session
+
+
+class TaskInstanceSchema(Schema):
+ """Task instance schema"""
+
+ task_id = fields.Str()
+ dag_id = fields.Str()
+ execution_date = fields.DateTime()
+ start_date = fields.DateTime()
+ end_date = fields.DateTime()
+ duration = fields.Float()
+ state = TaskInstanceStateField()
+ _try_number = fields.Int(data_key="try_number")
+ max_tries = fields.Int()
+ hostname = fields.Str()
+ unixname = fields.Str()
+ pool = fields.Str()
+ pool_slots = fields.Int()
+ queue = fields.Str()
+ priority_weight = fields.Int()
+ operator = fields.Str()
+ queued_dttm = fields.DateTime(data_key="queued_when")
+ pid = fields.Int()
+ executor_config = fields.Str()
+ sla_miss = fields.Method("get_sla_miss")
+
+ @staticmethod
+ def get_sla_miss(obj: TaskInstance):
+ with create_session() as session:
+ sla_miss = session.query(SlaMiss).filter(
Review comment:
Here we have an n+1 problem. I think we can fetch it much more efficiently if we use the more advanced features of SQLAlchemy.
https://docs.sqlalchemy.org/en/13/orm/loading_relationships.html
----------------------------------------------------------------
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] ashb commented on a change in pull request #9597: Add endpoints for task instances
Posted by GitBox <gi...@apache.org>.
ashb commented on a change in pull request #9597:
URL: https://github.com/apache/airflow/pull/9597#discussion_r503853886
##########
File path: airflow/api_connexion/endpoints/task_instance_endpoint.py
##########
@@ -14,23 +14,110 @@
# KIND, either express or implied. See the License for the
# specific language governing permissions and limitations
# under the License.
+from typing import Any, List, Optional, Tuple
-# TODO(mik-laj): We have to implement it.
-# Do you want to help? Please look at: https://github.com/apache/airflow/issues/8132
+from sqlalchemy import and_, func, or_
+from airflow.api_connexion.exceptions import NotFound
+from airflow.api_connexion.parameters import format_datetime, format_parameters
+from airflow.api_connexion.schemas.task_instance_schema import (
+ TaskInstanceCollection, task_instance_collection_schema, task_instance_schema,
+)
+from airflow.models.dagrun import DagRun as DR
+from airflow.models.taskinstance import TaskInstance as TI
+from airflow.utils.session import provide_session
+
+
+@provide_session
+def get_task_instance(dag_id: str, dag_run_id: str, task_id: str, session=None):
Review comment:
)Can be a follow up PR (and likely candidate would be to `TaskIstance.get_by_run_id()`)
----------------------------------------------------------------
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 #9597: [WIP] Add read-only endpoints for task instances
Posted by GitBox <gi...@apache.org>.
mik-laj commented on a change in pull request #9597:
URL: https://github.com/apache/airflow/pull/9597#discussion_r448590758
##########
File path: airflow/api_connexion/endpoints/task_instance_endpoint.py
##########
@@ -14,23 +14,110 @@
# KIND, either express or implied. See the License for the
# specific language governing permissions and limitations
# under the License.
+from typing import Any, List, Optional, Tuple
-# TODO(mik-laj): We have to implement it.
-# Do you want to help? Please look at: https://github.com/apache/airflow/issues/8132
+from sqlalchemy import and_, func, or_
+from airflow.api_connexion.exceptions import NotFound
+from airflow.api_connexion.parameters import format_datetime, format_parameters
+from airflow.api_connexion.schemas.task_instance_schema import (
+ TaskInstanceCollection, task_instance_collection_schema, task_instance_schema,
+)
+from airflow.models.dagrun import DagRun as DR
+from airflow.models.taskinstance import TaskInstance as TI
+from airflow.utils.session import provide_session
+
+
+@provide_session
+def get_task_instance(dag_id: str, dag_run_id: str, task_id: str, session=None):
+ query = (
+ session.query(TI)
+ .filter(TI.dag_id == dag_id)
+ .join(DR, and_(TI.dag_id == DR.dag_id, TI.execution_date == DR.execution_date))
+ .filter(DR.run_id == dag_run_id)
+ .filter(TI.task_id == task_id)
+ )
+
+ task_instance = query.one_or_none()
+
+ if task_instance is None:
+ raise NotFound("Task instance not found")
+
+ return task_instance_schema.dump(task_instance)
-def get_task_instance():
- """
- Get a task instance
- """
- raise NotImplementedError("Not implemented yet.")
+def _apply_array_filter(query, key, values):
+ if values is not None:
+ query = query.filter(or_(*[key == v for v in values]))
+ return query
-def get_task_instances():
+
+def _apply_range_filter(query, key, value_range: Tuple[Any, Any]):
+ gte_value, lte_value = value_range
+ if gte_value is not None:
+ query = query.filter(key >= gte_value)
+ if lte_value is not None:
+ query = query.filter(key <= lte_value)
+ return query
+
+
+@format_parameters(
+ {
+ 'start_date_gte': format_datetime,
+ 'start_date_lte': format_datetime,
+ 'execution_date_gte': format_datetime,
+ 'execution_date_lte': format_datetime,
+ 'end_date_gte': format_datetime,
+ 'end_date_lte': format_datetime,
+ }
+)
+@provide_session
+def get_task_instances(
+ limit: int,
+ dag_id: Optional[str] = None,
+ dag_run_id: Optional[str] = None,
+ execution_date_gte: Optional[str] = None,
+ execution_date_lte: Optional[str] = None,
+ start_date_gte: Optional[str] = None,
+ start_date_lte: Optional[str] = None,
+ end_date_gte: Optional[str] = None,
+ end_date_lte: Optional[str] = None,
+ duration_gte: Optional[float] = None,
+ duration_lte: Optional[float] = None,
+ state: Optional[str] = None,
+ pool: Optional[List[str]] = None,
+ queue: Optional[List[str]] = None,
+ offset: Optional[int] = None,
+ session=None,
Review comment:
It is not correct with pylint if I remember correctly. I didn't specify the type here because SQLAlchemy does a lot of magic, and mypy likes to get lost when he is aware of its existence.
----------------------------------------------------------------
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 #9597: Add endpoints for task instances
Posted by GitBox <gi...@apache.org>.
mik-laj commented on pull request #9597:
URL: https://github.com/apache/airflow/pull/9597#issuecomment-707575109
@turbaszek @kaxil @jhtimmins Can I ask for a 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] kaxil commented on a change in pull request #9597: Add endpoints for task instances
Posted by GitBox <gi...@apache.org>.
kaxil commented on a change in pull request #9597:
URL: https://github.com/apache/airflow/pull/9597#discussion_r503505201
##########
File path: airflow/api_connexion/endpoints/task_instance_endpoint.py
##########
@@ -14,46 +14,260 @@
# KIND, either express or implied. See the License for the
# specific language governing permissions and limitations
# under the License.
+from typing import Any, List, Optional, Tuple
+
+from flask import request
+from marshmallow import ValidationError
+from sqlalchemy import and_, func
-# TODO(mik-laj): We have to implement it.
-# Do you want to help? Please look at: https://github.com/apache/airflow/issues/8132
from airflow.api_connexion import security
+from airflow.api_connexion.exceptions import NotFound, BadRequest
+from airflow.api_connexion.parameters import format_datetime, format_parameters
+from airflow.api_connexion.schemas.task_instance_schema import (
+ clear_task_instance_form,
+ TaskInstanceCollection,
+ task_instance_collection_schema,
+ task_instance_schema,
+ task_instance_batch_form,
+ task_instance_reference_collection_schema,
+ TaskInstanceReferenceCollection,
+)
+from airflow.configuration import conf
+from airflow.models.dagrun import DagRun as DR
+from airflow.models.taskinstance import clear_task_instances, TaskInstance as TI
+from airflow.models.dag import DagModel
+from airflow.models.dagbag import DagBag
+from airflow.models import SlaMiss
+from airflow.utils.state import State
+from airflow.utils.session import provide_session
-@security.requires_access([("can_read", "Dag"), ("can_read", "DagRun"), ("can_read", "Task")])
-def get_task_instance():
- """
- Get a task instance
- """
- raise NotImplementedError("Not implemented yet.")
+@security.requires_access(
+ [
+ ("can_read", "Dag"),
+ ("can_read", "DagRun"),
+ ('can_read', 'Task'),
+ ]
+)
+@provide_session
+def get_task_instance(dag_id: str, dag_run_id: str, task_id: str, session=None):
+ """Get task instance"""
+ query = (
+ session.query(TI)
+ .filter(TI.dag_id == dag_id)
+ .join(DR, and_(TI.dag_id == DR.dag_id, TI.execution_date == DR.execution_date))
+ .filter(DR.run_id == dag_run_id)
+ .filter(TI.task_id == task_id)
+ .join(
+ SlaMiss,
+ and_(
+ SlaMiss.dag_id == TI.dag_id,
+ SlaMiss.execution_date == TI.execution_date,
+ SlaMiss.task_id == TI.task_id,
+ ),
+ isouter=True,
+ )
+ .add_entity(SlaMiss)
+ )
+ task_instance = query.one_or_none()
+ if task_instance is None:
+ raise NotFound("Task instance not found")
+ return task_instance_schema.dump(task_instance)
-@security.requires_access([("can_read", "Dag"), ("can_read", "DagRun"), ("can_read", "Task")])
-def get_task_instances():
- """
- Get list of task instances of DAG.
- """
- raise NotImplementedError("Not implemented yet.")
+
+def _apply_array_filter(query, key, values):
+ if values is not None:
+ query = query.filter(key.in_(values))
+ return query
+
+
+def _apply_range_filter(query, key, value_range: Tuple[Any, Any]):
+ gte_value, lte_value = value_range
+ if gte_value is not None:
+ query = query.filter(key >= gte_value)
+ if lte_value is not None:
+ query = query.filter(key <= lte_value)
+ return query
+@format_parameters(
+ {
+ "execution_date_gte": format_datetime,
+ "execution_date_lte": format_datetime,
+ "start_date_gte": format_datetime,
+ "start_date_lte": format_datetime,
+ "end_date_gte": format_datetime,
+ "end_date_lte": format_datetime,
+ }
+)
@security.requires_access(
[
- ('can_read', 'Dag'),
- ('can_read', 'DagRun'),
+ ("can_read", "Dag"),
+ ("can_read", "DagRun"),
('can_read', 'Task'),
- ('can_read', 'TaskInstance'),
]
)
-def get_task_instances_batch():
+@provide_session
+def get_task_instances(
+ limit: int,
+ dag_id: Optional[str] = None,
+ dag_run_id: Optional[str] = None,
+ execution_date_gte: Optional[str] = None,
+ execution_date_lte: Optional[str] = None,
+ start_date_gte: Optional[str] = None,
+ start_date_lte: Optional[str] = None,
+ end_date_gte: Optional[str] = None,
+ end_date_lte: Optional[str] = None,
+ duration_gte: Optional[float] = None,
+ duration_lte: Optional[float] = None,
+ state: Optional[str] = None,
+ pool: Optional[List[str]] = None,
+ queue: Optional[List[str]] = None,
+ offset: Optional[int] = None,
+ session=None,
+): # pylint: disable=too-many-arguments
"""
Get list of task instances.
"""
- raise NotImplementedError("Not implemented yet.")
+ base_query = session.query(TI)
+
+ if dag_id != "~":
+ base_query = base_query.filter(TI.dag_id == dag_id)
+ if dag_run_id != "~":
+ base_query = base_query.join(DR, and_(TI.dag_id == DR.dag_id, TI.execution_date == DR.execution_date))
Review comment:
same as https://github.com/apache/airflow/pull/9597/files#r503503456
----------------------------------------------------------------
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] github-actions[bot] commented on pull request #9597: Add endpoints for task instances
Posted by GitBox <gi...@apache.org>.
github-actions[bot] commented on pull request #9597:
URL: https://github.com/apache/airflow/pull/9597#issuecomment-706265181
[The Workflow run](https://github.com/apache/airflow/actions/runs/297799117) is cancelling this PR. It has some failed jobs matching ^Pylint$,^Static checks$,^Build docs$,^Spell check docs$,^Backport packages$,^Checks: Helm tests$,^Test OpenAPI*.
----------------------------------------------------------------
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] github-actions[bot] commented on pull request #9597: Add endpoints for task instances
Posted by GitBox <gi...@apache.org>.
github-actions[bot] commented on pull request #9597:
URL: https://github.com/apache/airflow/pull/9597#issuecomment-706169947
[The Workflow run](https://github.com/apache/airflow/actions/runs/297511179) is cancelling this PR. It has some failed jobs matching ^Pylint$,^Static checks$,^Build docs$,^Spell check docs$,^Backport packages$,^Checks: Helm tests$,^Test OpenAPI*.
----------------------------------------------------------------
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 #9597: Add endpoints for task instances
Posted by GitBox <gi...@apache.org>.
kaxil commented on a change in pull request #9597:
URL: https://github.com/apache/airflow/pull/9597#discussion_r503503456
##########
File path: airflow/api_connexion/endpoints/task_instance_endpoint.py
##########
@@ -14,46 +14,260 @@
# KIND, either express or implied. See the License for the
# specific language governing permissions and limitations
# under the License.
+from typing import Any, List, Optional, Tuple
+
+from flask import request
+from marshmallow import ValidationError
+from sqlalchemy import and_, func
-# TODO(mik-laj): We have to implement it.
-# Do you want to help? Please look at: https://github.com/apache/airflow/issues/8132
from airflow.api_connexion import security
+from airflow.api_connexion.exceptions import NotFound, BadRequest
+from airflow.api_connexion.parameters import format_datetime, format_parameters
+from airflow.api_connexion.schemas.task_instance_schema import (
+ clear_task_instance_form,
+ TaskInstanceCollection,
+ task_instance_collection_schema,
+ task_instance_schema,
+ task_instance_batch_form,
+ task_instance_reference_collection_schema,
+ TaskInstanceReferenceCollection,
+)
+from airflow.configuration import conf
+from airflow.models.dagrun import DagRun as DR
+from airflow.models.taskinstance import clear_task_instances, TaskInstance as TI
+from airflow.models.dag import DagModel
+from airflow.models.dagbag import DagBag
+from airflow.models import SlaMiss
+from airflow.utils.state import State
+from airflow.utils.session import provide_session
-@security.requires_access([("can_read", "Dag"), ("can_read", "DagRun"), ("can_read", "Task")])
-def get_task_instance():
- """
- Get a task instance
- """
- raise NotImplementedError("Not implemented yet.")
+@security.requires_access(
+ [
+ ("can_read", "Dag"),
+ ("can_read", "DagRun"),
+ ('can_read', 'Task'),
+ ]
+)
+@provide_session
+def get_task_instance(dag_id: str, dag_run_id: str, task_id: str, session=None):
+ """Get task instance"""
+ query = (
+ session.query(TI)
+ .filter(TI.dag_id == dag_id)
+ .join(DR, and_(TI.dag_id == DR.dag_id, TI.execution_date == DR.execution_date))
Review comment:
```suggestion
.join(DR)
```
should work because of https://github.com/apache/airflow/blob/c47383dffff1d216d1c4985b4a9cc7a7310a837e/airflow/models/dagrun.py#L77-L82
----------------------------------------------------------------
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 #9597: Add endpoints for task instances
Posted by GitBox <gi...@apache.org>.
mik-laj commented on a change in pull request #9597:
URL: https://github.com/apache/airflow/pull/9597#discussion_r502515103
##########
File path: airflow/api_connexion/endpoints/task_instance_endpoint.py
##########
@@ -14,39 +14,234 @@
# KIND, either express or implied. See the License for the
# specific language governing permissions and limitations
# under the License.
+from typing import Any, List, Optional, Tuple
+
+from flask import request
+from marshmallow import ValidationError
+from sqlalchemy import and_, func, or_
-# TODO(mik-laj): We have to implement it.
-# Do you want to help? Please look at: https://github.com/apache/airflow/issues/8132
from airflow.api_connexion import security
+from airflow.api_connexion.exceptions import NotFound, BadRequest
+from airflow.api_connexion.parameters import format_datetime, format_parameters
+from airflow.api_connexion.schemas.task_instance_schema import (
+ clear_task_instance_form,
+ TaskInstanceCollection,
+ task_instance_collection_schema,
+ task_instance_schema,
+ task_instance_batch_form,
+ task_instance_reference_collection_schema,
+ TaskInstanceReferenceCollection,
+)
+from airflow.configuration import conf
+from airflow.models.dagrun import DagRun as DR
+from airflow.models.taskinstance import clear_task_instances, TaskInstance as TI
+from airflow.models.dag import DagModel
+from airflow.models.dagbag import DagBag
+from airflow.models import SlaMiss
+from airflow.utils.state import State
+from airflow.utils.session import provide_session
@security.requires_authentication
-def get_task_instance():
- """
- Get a task instance
- """
- raise NotImplementedError("Not implemented yet.")
+@provide_session
+def get_task_instance(dag_id: str, dag_run_id: str, task_id: str, session=None):
+ """Get task instance"""
+ query = (
+ session.query(TI)
+ .filter(TI.dag_id == dag_id)
+ .join(DR, and_(TI.dag_id == DR.dag_id, TI.execution_date == DR.execution_date))
+ .filter(DR.run_id == dag_run_id)
+ .filter(TI.task_id == task_id)
+ .join(
+ SlaMiss,
+ and_(
+ SlaMiss.dag_id == TI.dag_id,
+ SlaMiss.execution_date == TI.execution_date,
+ SlaMiss.task_id == TI.task_id,
+ ),
+ isouter=True,
+ )
+ .add_entity(SlaMiss)
+ )
+ task_instance = query.one_or_none()
+
+ if task_instance is None:
+ raise NotFound("Task instance not found")
+
+ return task_instance_schema.dump(task_instance)
+
+
+def _apply_array_filter(query, key, values):
+ if values is not None:
+ query = query.filter(or_(*[key == v for v in values]))
+ return query
+
+
+def _apply_range_filter(query, key, value_range: Tuple[Any, Any]):
+ gte_value, lte_value = value_range
+ if gte_value is not None:
+ query = query.filter(key >= gte_value)
+ if lte_value is not None:
+ query = query.filter(key <= lte_value)
+ return query
+@format_parameters(
+ {
+ "execution_date_gte": format_datetime,
+ "execution_date_lte": format_datetime,
+ "start_date_gte": format_datetime,
+ "start_date_lte": format_datetime,
+ "end_date_gte": format_datetime,
+ "end_date_lte": format_datetime,
+ }
+)
@security.requires_authentication
-def get_task_instances():
+@provide_session
+def get_task_instances(
+ limit: int,
+ dag_id: Optional[str] = None,
+ dag_run_id: Optional[str] = None,
+ execution_date_gte: Optional[str] = None,
+ execution_date_lte: Optional[str] = None,
+ start_date_gte: Optional[str] = None,
+ start_date_lte: Optional[str] = None,
+ end_date_gte: Optional[str] = None,
+ end_date_lte: Optional[str] = None,
+ duration_gte: Optional[float] = None,
+ duration_lte: Optional[float] = None,
+ state: Optional[str] = None,
+ pool: Optional[List[str]] = None,
+ queue: Optional[List[str]] = None,
+ offset: Optional[int] = None,
+ session=None,
+): # pylint: disable=too-many-arguments
"""
- Get list of task instances of DAG.
+ Get list of task instances.
"""
- raise NotImplementedError("Not implemented yet.")
+ query = session.query(TI)
+
+ if dag_id != "~":
+ query = query.filter(TI.dag_id == dag_id)
+ if dag_run_id != "~":
+ query = query.join(DR, and_(TI.dag_id == DR.dag_id, TI.execution_date == DR.execution_date))
+ query = query.filter(DR.run_id == dag_run_id)
+ query = _apply_range_filter(
+ query, key=DR.execution_date, value_range=(execution_date_gte, execution_date_lte),
+ )
+ query = _apply_range_filter(query, key=TI.start_date, value_range=(start_date_gte, start_date_lte))
+ query = _apply_range_filter(query, key=TI.end_date, value_range=(end_date_gte, end_date_lte))
+ query = _apply_range_filter(query, key=TI.duration, value_range=(duration_gte, duration_lte))
+ query = _apply_array_filter(query, key=TI.state, values=state)
+ query = _apply_array_filter(query, key=TI.pool, values=pool)
+ query = _apply_array_filter(query, key=TI.queue, values=queue)
+
+ query = query.join(
+ SlaMiss,
+ and_(
+ SlaMiss.dag_id == TI.dag_id,
+ SlaMiss.task_id == TI.task_id,
+ SlaMiss.execution_date == TI.execution_date,
+ ),
+ isouter=True,
+ ).add_entity(SlaMiss)
+ task_instances = query.offset(offset).limit(limit).all()
+ total_entries = session.query(
+ func.count(func.distinct(TI.dag_id, TI.execution_date, TI.task_id))
+ ).scalar()
+ return task_instance_collection_schema.dump(
+ TaskInstanceCollection(task_instances=task_instances, total_entries=total_entries)
+ )
@security.requires_authentication
-def get_task_instances_batch():
+@provide_session
+def get_task_instances_batch(session=None):
"""
Get list of task instances.
"""
- raise NotImplementedError("Not implemented yet.")
+ body = request.get_json()
+ try:
+ data = task_instance_batch_form.load(body)
+ except ValidationError as err:
+ raise BadRequest(detail=str(err.messages))
+ query = session.query(TI)
+
+ query = _apply_array_filter(query, key=TI.dag_id, values=data["dag_ids"])
+ query = _apply_range_filter(
+ query, key=TI.execution_date, value_range=(data["execution_date_gte"], data["execution_date_lte"]),
+ )
+ query = _apply_range_filter(
+ query, key=TI.start_date, value_range=(data["start_date_gte"], data["start_date_lte"]),
+ )
+ query = _apply_range_filter(
+ query, key=TI.end_date, value_range=(data["end_date_gte"], data["end_date_lte"])
+ )
+ query = _apply_range_filter(
+ query, key=TI.duration, value_range=(data["duration_gte"], data["duration_lte"])
+ )
+ query = _apply_array_filter(query, key=TI.state, values=data["state"])
+ query = _apply_array_filter(query, key=TI.pool, values=data["pool"])
+ query = _apply_array_filter(query, key=TI.queue, values=data["queue"])
+
+ query = query.join(
+ SlaMiss,
+ and_(
+ SlaMiss.dag_id == TI.dag_id,
+ SlaMiss.task_id == TI.task_id,
+ SlaMiss.execution_date == TI.execution_date,
+ ),
+ isouter=True,
+ ).add_entity(SlaMiss)
Review comment:
LGTM.
----------------------------------------------------------------
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 #9597: [WIP] Add read-only endpoints for task instances
Posted by GitBox <gi...@apache.org>.
kaxil commented on a change in pull request #9597:
URL: https://github.com/apache/airflow/pull/9597#discussion_r489338216
##########
File path: airflow/api_connexion/endpoints/task_instance_endpoint.py
##########
@@ -14,39 +14,234 @@
# KIND, either express or implied. See the License for the
# specific language governing permissions and limitations
# under the License.
+from typing import Any, List, Optional, Tuple
+
+from flask import request
+from marshmallow import ValidationError
+from sqlalchemy import and_, func, or_
-# TODO(mik-laj): We have to implement it.
-# Do you want to help? Please look at: https://github.com/apache/airflow/issues/8132
from airflow.api_connexion import security
+from airflow.api_connexion.exceptions import NotFound, BadRequest
+from airflow.api_connexion.parameters import format_datetime, format_parameters
+from airflow.api_connexion.schemas.task_instance_schema import (
+ clear_task_instance_form,
+ TaskInstanceCollection,
+ task_instance_collection_schema,
+ task_instance_schema,
+ task_instance_batch_form,
+ task_instance_reference_collection_schema,
+ TaskInstanceReferenceCollection,
+)
+from airflow.configuration import conf
+from airflow.models.dagrun import DagRun as DR
+from airflow.models.taskinstance import clear_task_instances, TaskInstance as TI
+from airflow.models.dag import DagModel
+from airflow.models.dagbag import DagBag
+from airflow.models import SlaMiss
+from airflow.utils.state import State
+from airflow.utils.session import provide_session
@security.requires_authentication
-def get_task_instance():
- """
- Get a task instance
- """
- raise NotImplementedError("Not implemented yet.")
+@provide_session
+def get_task_instance(dag_id: str, dag_run_id: str, task_id: str, session=None):
+ """Get task instance"""
+ query = (
+ session.query(TI)
+ .filter(TI.dag_id == dag_id)
+ .join(DR, and_(TI.dag_id == DR.dag_id, TI.execution_date == DR.execution_date))
+ .filter(DR.run_id == dag_run_id)
+ .filter(TI.task_id == task_id)
+ .join(
+ SlaMiss,
+ and_(
+ SlaMiss.dag_id == TI.dag_id,
+ SlaMiss.execution_date == TI.execution_date,
+ SlaMiss.task_id == TI.task_id,
+ ),
+ isouter=True,
+ )
+ .add_entity(SlaMiss)
+ )
+ task_instance = query.one_or_none()
+
+ if task_instance is None:
+ raise NotFound("Task instance not found")
+
+ return task_instance_schema.dump(task_instance)
+
+
+def _apply_array_filter(query, key, values):
+ if values is not None:
+ query = query.filter(or_(*[key == v for v in values]))
+ return query
+
+
+def _apply_range_filter(query, key, value_range: Tuple[Any, Any]):
+ gte_value, lte_value = value_range
+ if gte_value is not None:
+ query = query.filter(key >= gte_value)
+ if lte_value is not None:
+ query = query.filter(key <= lte_value)
+ return query
+@format_parameters(
+ {
+ "execution_date_gte": format_datetime,
+ "execution_date_lte": format_datetime,
+ "start_date_gte": format_datetime,
+ "start_date_lte": format_datetime,
+ "end_date_gte": format_datetime,
+ "end_date_lte": format_datetime,
+ }
+)
@security.requires_authentication
-def get_task_instances():
+@provide_session
+def get_task_instances(
+ limit: int,
+ dag_id: Optional[str] = None,
+ dag_run_id: Optional[str] = None,
+ execution_date_gte: Optional[str] = None,
+ execution_date_lte: Optional[str] = None,
+ start_date_gte: Optional[str] = None,
+ start_date_lte: Optional[str] = None,
+ end_date_gte: Optional[str] = None,
+ end_date_lte: Optional[str] = None,
+ duration_gte: Optional[float] = None,
+ duration_lte: Optional[float] = None,
+ state: Optional[str] = None,
+ pool: Optional[List[str]] = None,
+ queue: Optional[List[str]] = None,
+ offset: Optional[int] = None,
+ session=None,
+): # pylint: disable=too-many-arguments
"""
- Get list of task instances of DAG.
+ Get list of task instances.
"""
- raise NotImplementedError("Not implemented yet.")
+ query = session.query(TI)
+
+ if dag_id != "~":
+ query = query.filter(TI.dag_id == dag_id)
+ if dag_run_id != "~":
+ query = query.join(DR, and_(TI.dag_id == DR.dag_id, TI.execution_date == DR.execution_date))
+ query = query.filter(DR.run_id == dag_run_id)
+ query = _apply_range_filter(
+ query, key=DR.execution_date, value_range=(execution_date_gte, execution_date_lte),
+ )
+ query = _apply_range_filter(query, key=TI.start_date, value_range=(start_date_gte, start_date_lte))
+ query = _apply_range_filter(query, key=TI.end_date, value_range=(end_date_gte, end_date_lte))
+ query = _apply_range_filter(query, key=TI.duration, value_range=(duration_gte, duration_lte))
+ query = _apply_array_filter(query, key=TI.state, values=state)
+ query = _apply_array_filter(query, key=TI.pool, values=pool)
+ query = _apply_array_filter(query, key=TI.queue, values=queue)
+
+ query = query.join(
+ SlaMiss,
+ and_(
+ SlaMiss.dag_id == TI.dag_id,
+ SlaMiss.task_id == TI.task_id,
+ SlaMiss.execution_date == TI.execution_date,
+ ),
+ isouter=True,
+ ).add_entity(SlaMiss)
+ task_instances = query.offset(offset).limit(limit).all()
+ total_entries = session.query(
+ func.count(func.distinct(TI.dag_id, TI.execution_date, TI.task_id))
+ ).scalar()
+ return task_instance_collection_schema.dump(
+ TaskInstanceCollection(task_instances=task_instances, total_entries=total_entries)
+ )
@security.requires_authentication
-def get_task_instances_batch():
+@provide_session
+def get_task_instances_batch(session=None):
"""
Get list of task instances.
"""
- raise NotImplementedError("Not implemented yet.")
+ body = request.get_json()
+ try:
+ data = task_instance_batch_form.load(body)
+ except ValidationError as err:
+ raise BadRequest(detail=str(err.messages))
+ query = session.query(TI)
+
+ query = _apply_array_filter(query, key=TI.dag_id, values=data["dag_ids"])
+ query = _apply_range_filter(
+ query, key=TI.execution_date, value_range=(data["execution_date_gte"], data["execution_date_lte"]),
+ )
+ query = _apply_range_filter(
+ query, key=TI.start_date, value_range=(data["start_date_gte"], data["start_date_lte"]),
+ )
+ query = _apply_range_filter(
+ query, key=TI.end_date, value_range=(data["end_date_gte"], data["end_date_lte"])
+ )
+ query = _apply_range_filter(
+ query, key=TI.duration, value_range=(data["duration_gte"], data["duration_lte"])
+ )
+ query = _apply_array_filter(query, key=TI.state, values=data["state"])
+ query = _apply_array_filter(query, key=TI.pool, values=data["pool"])
+ query = _apply_array_filter(query, key=TI.queue, values=data["queue"])
+
+ query = query.join(
+ SlaMiss,
+ and_(
+ SlaMiss.dag_id == TI.dag_id,
+ SlaMiss.task_id == TI.task_id,
+ SlaMiss.execution_date == TI.execution_date,
+ ),
+ isouter=True,
+ ).add_entity(SlaMiss)
+ task_instances = query.all()
+ total_entries = session.query(
+ func.count(func.distinct(TI.dag_id, TI.execution_date, TI.task_id))
+ ).scalar()
+
+ return task_instance_collection_schema.dump(
+ TaskInstanceCollection(task_instances=task_instances, total_entries=total_entries)
+ )
@security.requires_authentication
-def post_clear_task_instances():
+@provide_session
+def post_clear_task_instances(dag_id: str, session=None):
Review comment:
@mik-laj Is there currently no way in the new API to make a TI as success / failed ?
----------------------------------------------------------------
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 #9597: Add endpoints for task instances
Posted by GitBox <gi...@apache.org>.
kaxil commented on a change in pull request #9597:
URL: https://github.com/apache/airflow/pull/9597#discussion_r503505097
##########
File path: airflow/api_connexion/endpoints/task_instance_endpoint.py
##########
@@ -14,46 +14,260 @@
# KIND, either express or implied. See the License for the
# specific language governing permissions and limitations
# under the License.
+from typing import Any, List, Optional, Tuple
+
+from flask import request
+from marshmallow import ValidationError
+from sqlalchemy import and_, func
-# TODO(mik-laj): We have to implement it.
-# Do you want to help? Please look at: https://github.com/apache/airflow/issues/8132
from airflow.api_connexion import security
+from airflow.api_connexion.exceptions import NotFound, BadRequest
+from airflow.api_connexion.parameters import format_datetime, format_parameters
+from airflow.api_connexion.schemas.task_instance_schema import (
+ clear_task_instance_form,
+ TaskInstanceCollection,
+ task_instance_collection_schema,
+ task_instance_schema,
+ task_instance_batch_form,
+ task_instance_reference_collection_schema,
+ TaskInstanceReferenceCollection,
+)
+from airflow.configuration import conf
+from airflow.models.dagrun import DagRun as DR
+from airflow.models.taskinstance import clear_task_instances, TaskInstance as TI
+from airflow.models.dag import DagModel
+from airflow.models.dagbag import DagBag
+from airflow.models import SlaMiss
+from airflow.utils.state import State
+from airflow.utils.session import provide_session
-@security.requires_access([("can_read", "Dag"), ("can_read", "DagRun"), ("can_read", "Task")])
-def get_task_instance():
- """
- Get a task instance
- """
- raise NotImplementedError("Not implemented yet.")
+@security.requires_access(
+ [
+ ("can_read", "Dag"),
+ ("can_read", "DagRun"),
+ ('can_read', 'Task'),
+ ]
+)
+@provide_session
+def get_task_instance(dag_id: str, dag_run_id: str, task_id: str, session=None):
+ """Get task instance"""
+ query = (
+ session.query(TI)
+ .filter(TI.dag_id == dag_id)
+ .join(DR, and_(TI.dag_id == DR.dag_id, TI.execution_date == DR.execution_date))
+ .filter(DR.run_id == dag_run_id)
+ .filter(TI.task_id == task_id)
+ .join(
+ SlaMiss,
+ and_(
+ SlaMiss.dag_id == TI.dag_id,
+ SlaMiss.execution_date == TI.execution_date,
+ SlaMiss.task_id == TI.task_id,
+ ),
+ isouter=True,
+ )
+ .add_entity(SlaMiss)
+ )
+ task_instance = query.one_or_none()
+ if task_instance is None:
+ raise NotFound("Task instance not found")
+ return task_instance_schema.dump(task_instance)
-@security.requires_access([("can_read", "Dag"), ("can_read", "DagRun"), ("can_read", "Task")])
-def get_task_instances():
- """
- Get list of task instances of DAG.
- """
- raise NotImplementedError("Not implemented yet.")
+
+def _apply_array_filter(query, key, values):
+ if values is not None:
+ query = query.filter(key.in_(values))
+ return query
+
+
+def _apply_range_filter(query, key, value_range: Tuple[Any, Any]):
+ gte_value, lte_value = value_range
+ if gte_value is not None:
+ query = query.filter(key >= gte_value)
+ if lte_value is not None:
+ query = query.filter(key <= lte_value)
+ return query
+@format_parameters(
+ {
+ "execution_date_gte": format_datetime,
+ "execution_date_lte": format_datetime,
+ "start_date_gte": format_datetime,
+ "start_date_lte": format_datetime,
+ "end_date_gte": format_datetime,
+ "end_date_lte": format_datetime,
+ }
+)
@security.requires_access(
[
- ('can_read', 'Dag'),
- ('can_read', 'DagRun'),
+ ("can_read", "Dag"),
+ ("can_read", "DagRun"),
('can_read', 'Task'),
- ('can_read', 'TaskInstance'),
]
)
-def get_task_instances_batch():
+@provide_session
+def get_task_instances(
+ limit: int,
+ dag_id: Optional[str] = None,
+ dag_run_id: Optional[str] = None,
+ execution_date_gte: Optional[str] = None,
+ execution_date_lte: Optional[str] = None,
+ start_date_gte: Optional[str] = None,
+ start_date_lte: Optional[str] = None,
+ end_date_gte: Optional[str] = None,
+ end_date_lte: Optional[str] = None,
+ duration_gte: Optional[float] = None,
+ duration_lte: Optional[float] = None,
+ state: Optional[str] = None,
+ pool: Optional[List[str]] = None,
+ queue: Optional[List[str]] = None,
+ offset: Optional[int] = None,
+ session=None,
+): # pylint: disable=too-many-arguments
"""
Get list of task instances.
"""
- raise NotImplementedError("Not implemented yet.")
+ base_query = session.query(TI)
+
+ if dag_id != "~":
+ base_query = base_query.filter(TI.dag_id == dag_id)
+ if dag_run_id != "~":
+ base_query = base_query.join(DR, and_(TI.dag_id == DR.dag_id, TI.execution_date == DR.execution_date))
Review comment:
```suggestion
base_query = base_query.join(DR)
```
----------------------------------------------------------------
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 #9597: [WIP] Add read-only endpoints for task instances
Posted by GitBox <gi...@apache.org>.
mik-laj commented on a change in pull request #9597:
URL: https://github.com/apache/airflow/pull/9597#discussion_r448588204
##########
File path: airflow/api_connexion/endpoints/task_instance_endpoint.py
##########
@@ -14,23 +14,110 @@
# KIND, either express or implied. See the License for the
# specific language governing permissions and limitations
# under the License.
+from typing import Any, List, Optional, Tuple
-# TODO(mik-laj): We have to implement it.
-# Do you want to help? Please look at: https://github.com/apache/airflow/issues/8132
+from sqlalchemy import and_, func, or_
+from airflow.api_connexion.exceptions import NotFound
+from airflow.api_connexion.parameters import format_datetime, format_parameters
+from airflow.api_connexion.schemas.task_instance_schema import (
+ TaskInstanceCollection, task_instance_collection_schema, task_instance_schema,
+)
+from airflow.models.dagrun import DagRun as DR
+from airflow.models.taskinstance import TaskInstance as TI
+from airflow.utils.session import provide_session
+
+
+@provide_session
+def get_task_instance(dag_id: str, dag_run_id: str, task_id: str, session=None):
+ query = (
+ session.query(TI)
+ .filter(TI.dag_id == dag_id)
+ .join(DR, and_(TI.dag_id == DR.dag_id, TI.execution_date == DR.execution_date))
+ .filter(DR.run_id == dag_run_id)
+ .filter(TI.task_id == task_id)
+ )
+
+ task_instance = query.one_or_none()
+
+ if task_instance is None:
+ raise NotFound("Task instance not found")
+
+ return task_instance_schema.dump(task_instance)
-def get_task_instance():
- """
- Get a task instance
- """
- raise NotImplementedError("Not implemented yet.")
+def _apply_array_filter(query, key, values):
+ if values is not None:
+ query = query.filter(or_(*[key == v for v in values]))
+ return query
-def get_task_instances():
+
+def _apply_range_filter(query, key, value_range: Tuple[Any, Any]):
+ gte_value, lte_value = value_range
+ if gte_value is not None:
+ query = query.filter(key >= gte_value)
+ if lte_value is not None:
+ query = query.filter(key <= lte_value)
+ return query
+
+
+@format_parameters(
+ {
+ 'start_date_gte': format_datetime,
+ 'start_date_lte': format_datetime,
+ 'execution_date_gte': format_datetime,
+ 'execution_date_lte': format_datetime,
+ 'end_date_gte': format_datetime,
+ 'end_date_lte': format_datetime,
+ }
+)
+@provide_session
+def get_task_instances(
+ limit: int,
+ dag_id: Optional[str] = None,
+ dag_run_id: Optional[str] = None,
+ execution_date_gte: Optional[str] = None,
+ execution_date_lte: Optional[str] = None,
+ start_date_gte: Optional[str] = None,
+ start_date_lte: Optional[str] = None,
+ end_date_gte: Optional[str] = None,
+ end_date_lte: Optional[str] = None,
+ duration_gte: Optional[float] = None,
+ duration_lte: Optional[float] = None,
+ state: Optional[str] = None,
+ pool: Optional[List[str]] = None,
+ queue: Optional[List[str]] = None,
+ offset: Optional[int] = None,
+ session=None,
+):
"""
- Get list of task instances of DAG.
+ Get list of a task instances
"""
- raise NotImplementedError("Not implemented yet.")
+ query = session.query(TI)
+
+ if dag_id is not None:
+ query = query.filter(TI.dag_id == dag_id)
+ if dag_run_id is not None:
+ query = query.join(DR, and_(TI.dag_id == DR.dag_id, TI.execution_date == DR.execution_date))
+ query = query.filter(DR.run_id == dag_run_id)
+
+ query = _apply_range_filter(
+ query, key=DR.execution_date, value_range=(execution_date_gte, execution_date_lte)
+ )
+ query = _apply_range_filter(query, key=DR.start_date, value_range=(start_date_gte, start_date_lte))
+ query = _apply_range_filter(query, key=DR.end_date, value_range=(end_date_gte, end_date_lte))
+ query = _apply_range_filter(query, key=DR.end_date, value_range=(end_date_gte, end_date_lte))
Review comment:
Good point. Itt should be execution_date.
----------------------------------------------------------------
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 #9597: [WIP] Add read-only endpoints for task instances
Posted by GitBox <gi...@apache.org>.
mik-laj commented on a change in pull request #9597:
URL: https://github.com/apache/airflow/pull/9597#discussion_r478797281
##########
File path: airflow/api_connexion/schemas/task_instance_schema.py
##########
@@ -0,0 +1,147 @@
+# 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, ValidationError, post_load
+from marshmallow_sqlalchemy import SQLAlchemySchema, auto_field
+from sqlalchemy import and_
+
+from airflow.api_connexion.schemas.enum_schemas import TaskInstanceStateField
+from airflow.api_connexion.schemas.sla_miss_schema import SlaMissSchema
+from airflow.models import DagRun, TaskInstance, SlaMiss
+from airflow.utils.session import create_session
+
+
+class TaskInstanceSchema(Schema):
+ """Task instance schema"""
+
+ task_id = fields.Str()
+ dag_id = fields.Str()
+ execution_date = fields.DateTime()
+ start_date = fields.DateTime()
+ end_date = fields.DateTime()
+ duration = fields.Float()
+ state = TaskInstanceStateField()
+ _try_number = fields.Int(data_key="try_number")
+ max_tries = fields.Int()
+ hostname = fields.Str()
+ unixname = fields.Str()
+ pool = fields.Str()
+ pool_slots = fields.Int()
+ queue = fields.Str()
+ priority_weight = fields.Int()
+ operator = fields.Str()
+ queued_dttm = fields.DateTime(data_key="queued_when")
+ pid = fields.Int()
+ executor_config = fields.Str()
+ sla_miss = fields.Method("get_sla_miss")
+
+ @staticmethod
+ def get_sla_miss(obj: TaskInstance):
+ with create_session() as session:
+ sla_miss = session.query(SlaMiss).filter(
Review comment:
I'll try to look at it tomorrow.
----------------------------------------------------------------
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 #9597: Add endpoints for task instances
Posted by GitBox <gi...@apache.org>.
mik-laj commented on a change in pull request #9597:
URL: https://github.com/apache/airflow/pull/9597#discussion_r503755336
##########
File path: airflow/api_connexion/endpoints/task_instance_endpoint.py
##########
@@ -14,46 +14,260 @@
# KIND, either express or implied. See the License for the
# specific language governing permissions and limitations
# under the License.
+from typing import Any, List, Optional, Tuple
+
+from flask import request
+from marshmallow import ValidationError
+from sqlalchemy import and_, func
-# TODO(mik-laj): We have to implement it.
-# Do you want to help? Please look at: https://github.com/apache/airflow/issues/8132
from airflow.api_connexion import security
+from airflow.api_connexion.exceptions import NotFound, BadRequest
+from airflow.api_connexion.parameters import format_datetime, format_parameters
+from airflow.api_connexion.schemas.task_instance_schema import (
+ clear_task_instance_form,
+ TaskInstanceCollection,
+ task_instance_collection_schema,
+ task_instance_schema,
+ task_instance_batch_form,
+ task_instance_reference_collection_schema,
+ TaskInstanceReferenceCollection,
+)
+from airflow.configuration import conf
+from airflow.models.dagrun import DagRun as DR
+from airflow.models.taskinstance import clear_task_instances, TaskInstance as TI
+from airflow.models.dag import DagModel
+from airflow.models.dagbag import DagBag
+from airflow.models import SlaMiss
+from airflow.utils.state import State
+from airflow.utils.session import provide_session
-@security.requires_access([("can_read", "Dag"), ("can_read", "DagRun"), ("can_read", "Task")])
-def get_task_instance():
- """
- Get a task instance
- """
- raise NotImplementedError("Not implemented yet.")
+@security.requires_access(
+ [
+ ("can_read", "Dag"),
+ ("can_read", "DagRun"),
+ ('can_read', 'Task'),
+ ]
+)
+@provide_session
+def get_task_instance(dag_id: str, dag_run_id: str, task_id: str, session=None):
+ """Get task instance"""
+ query = (
+ session.query(TI)
+ .filter(TI.dag_id == dag_id)
+ .join(DR, and_(TI.dag_id == DR.dag_id, TI.execution_date == DR.execution_date))
Review comment:
Unfortunately it doesn't work. I have the following message.
```
sqlalchemy.exc.InvalidRequestError: Don't know how to join to <class 'airflow.models.dagrun.DagRun'>. Please use the .select_from() method to establish an explicit left side, as well as providing an explcit ON clause if not present already to help resolve the ambiguity.
```
----------------------------------------------------------------
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 #9597: [WIP] Add read-only endpoints for task instances
Posted by GitBox <gi...@apache.org>.
mik-laj commented on a change in pull request #9597:
URL: https://github.com/apache/airflow/pull/9597#discussion_r448587966
##########
File path: airflow/api_connexion/endpoints/task_instance_endpoint.py
##########
@@ -14,23 +14,110 @@
# KIND, either express or implied. See the License for the
# specific language governing permissions and limitations
# under the License.
+from typing import Any, List, Optional, Tuple
-# TODO(mik-laj): We have to implement it.
-# Do you want to help? Please look at: https://github.com/apache/airflow/issues/8132
+from sqlalchemy import and_, func, or_
+from airflow.api_connexion.exceptions import NotFound
+from airflow.api_connexion.parameters import format_datetime, format_parameters
+from airflow.api_connexion.schemas.task_instance_schema import (
+ TaskInstanceCollection, task_instance_collection_schema, task_instance_schema,
+)
+from airflow.models.dagrun import DagRun as DR
+from airflow.models.taskinstance import TaskInstance as TI
+from airflow.utils.session import provide_session
+
+
+@provide_session
+def get_task_instance(dag_id: str, dag_run_id: str, task_id: str, session=None):
+ query = (
+ session.query(TI)
+ .filter(TI.dag_id == dag_id)
+ .join(DR, and_(TI.dag_id == DR.dag_id, TI.execution_date == DR.execution_date))
+ .filter(DR.run_id == dag_run_id)
+ .filter(TI.task_id == task_id)
+ )
+
+ task_instance = query.one_or_none()
+
+ if task_instance is None:
+ raise NotFound("Task instance not found")
+
+ return task_instance_schema.dump(task_instance)
-def get_task_instance():
- """
- Get a task instance
- """
- raise NotImplementedError("Not implemented yet.")
+def _apply_array_filter(query, key, values):
+ if values is not None:
+ query = query.filter(or_(*[key == v for v in values]))
+ return query
-def get_task_instances():
+
+def _apply_range_filter(query, key, value_range: Tuple[Any, Any]):
+ gte_value, lte_value = value_range
+ if gte_value is not None:
+ query = query.filter(key >= gte_value)
+ if lte_value is not None:
+ query = query.filter(key <= lte_value)
+ return query
+
+
+@format_parameters(
+ {
+ 'start_date_gte': format_datetime,
+ 'start_date_lte': format_datetime,
+ 'execution_date_gte': format_datetime,
+ 'execution_date_lte': format_datetime,
+ 'end_date_gte': format_datetime,
+ 'end_date_lte': format_datetime,
+ }
+)
+@provide_session
+def get_task_instances(
+ limit: int,
+ dag_id: Optional[str] = None,
+ dag_run_id: Optional[str] = None,
+ execution_date_gte: Optional[str] = None,
+ execution_date_lte: Optional[str] = None,
+ start_date_gte: Optional[str] = None,
+ start_date_lte: Optional[str] = None,
+ end_date_gte: Optional[str] = None,
+ end_date_lte: Optional[str] = None,
+ duration_gte: Optional[float] = None,
+ duration_lte: Optional[float] = None,
+ state: Optional[str] = None,
+ pool: Optional[List[str]] = None,
+ queue: Optional[List[str]] = None,
+ offset: Optional[int] = None,
+ session=None,
Review comment:
We use connexion, which fills these parameters based on the API specification.
https://connexion.readthedocs.io/en/latest/request.html#automatic-parameter-handling
----------------------------------------------------------------
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 #9597: Add endpoints for task instances
Posted by GitBox <gi...@apache.org>.
kaxil commented on a change in pull request #9597:
URL: https://github.com/apache/airflow/pull/9597#discussion_r503853508
##########
File path: airflow/api_connexion/openapi/v1.yaml
##########
@@ -2065,6 +2101,48 @@ components:
description: Set state of DAG runs to RUNNING.
type: boolean
+ UpdateTaskInstancesState:
+ type: object
+ properties:
+ dry_run:
+ description: |
+ If set, don't actually run this operation. The response will contain a list of task instances
+ planned to be affect, but not modified in any way.
Review comment:
```suggestion
If set, don't actually run this operation. The response will contain a list of task instances
planned to be affected, but won't be modified in any way.
```
----------------------------------------------------------------
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 #9597: [WIP] Add read-only endpoints for task instances
Posted by GitBox <gi...@apache.org>.
mik-laj commented on pull request #9597:
URL: https://github.com/apache/airflow/pull/9597#issuecomment-706051780
Everything works. I am merging all commits into one commit on my branch and doing rebase now
----------------------------------------------------------------
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 pull request #9597: [WIP] Add read-only endpoints for task instances
Posted by GitBox <gi...@apache.org>.
OmairK commented on pull request #9597:
URL: https://github.com/apache/airflow/pull/9597#issuecomment-703751945
> @OmairK @mik-laj Any updates here?
Waiting for @mik-laj's review on the approach I followed,
----------------------------------------------------------------
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 edited a comment on pull request #9597: [WIP] Add read-only endpoints for task instances
Posted by GitBox <gi...@apache.org>.
OmairK edited a comment on pull request #9597:
URL: https://github.com/apache/airflow/pull/9597#issuecomment-703751945
> @OmairK @mik-laj Any updates here?
Waiting for @mik-laj's review on the approach I followed.
----------------------------------------------------------------
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 #9597: [WIP] Add read-only endpoints for task instances
Posted by GitBox <gi...@apache.org>.
mik-laj commented on a change in pull request #9597:
URL: https://github.com/apache/airflow/pull/9597#discussion_r448586807
##########
File path: airflow/api_connexion/endpoints/task_instance_endpoint.py
##########
@@ -14,23 +14,110 @@
# KIND, either express or implied. See the License for the
# specific language governing permissions and limitations
# under the License.
+from typing import Any, List, Optional, Tuple
-# TODO(mik-laj): We have to implement it.
-# Do you want to help? Please look at: https://github.com/apache/airflow/issues/8132
+from sqlalchemy import and_, func, or_
+from airflow.api_connexion.exceptions import NotFound
+from airflow.api_connexion.parameters import format_datetime, format_parameters
+from airflow.api_connexion.schemas.task_instance_schema import (
+ TaskInstanceCollection, task_instance_collection_schema, task_instance_schema,
+)
+from airflow.models.dagrun import DagRun as DR
+from airflow.models.taskinstance import TaskInstance as TI
+from airflow.utils.session import provide_session
+
+
+@provide_session
+def get_task_instance(dag_id: str, dag_run_id: str, task_id: str, session=None):
+ query = (
+ session.query(TI)
+ .filter(TI.dag_id == dag_id)
+ .join(DR, and_(TI.dag_id == DR.dag_id, TI.execution_date == DR.execution_date))
+ .filter(DR.run_id == dag_run_id)
+ .filter(TI.task_id == task_id)
+ )
+
+ task_instance = query.one_or_none()
+
+ if task_instance is None:
+ raise NotFound("Task instance not found")
Review comment:
Yes. This is handled by connexion.
https://connexion.readthedocs.io/en/latest/exceptions.html
----------------------------------------------------------------
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] jhtimmins commented on a change in pull request #9597: [WIP] Add read-only endpoints for task instances
Posted by GitBox <gi...@apache.org>.
jhtimmins commented on a change in pull request #9597:
URL: https://github.com/apache/airflow/pull/9597#discussion_r451266466
##########
File path: airflow/api_connexion/endpoints/task_instance_endpoint.py
##########
@@ -14,23 +14,110 @@
# KIND, either express or implied. See the License for the
# specific language governing permissions and limitations
# under the License.
+from typing import Any, List, Optional, Tuple
-# TODO(mik-laj): We have to implement it.
-# Do you want to help? Please look at: https://github.com/apache/airflow/issues/8132
+from sqlalchemy import and_, func, or_
+from airflow.api_connexion.exceptions import NotFound
+from airflow.api_connexion.parameters import format_datetime, format_parameters
+from airflow.api_connexion.schemas.task_instance_schema import (
+ TaskInstanceCollection, task_instance_collection_schema, task_instance_schema,
+)
+from airflow.models.dagrun import DagRun as DR
+from airflow.models.taskinstance import TaskInstance as TI
+from airflow.utils.session import provide_session
+
+
+@provide_session
+def get_task_instance(dag_id: str, dag_run_id: str, task_id: str, session=None):
Review comment:
Ah sorry for being unclear. I was suggesting `get_task_instance` be moved to a different file outside the API.
----------------------------------------------------------------
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 #9597: Add endpoints for task instances
Posted by GitBox <gi...@apache.org>.
mik-laj commented on a change in pull request #9597:
URL: https://github.com/apache/airflow/pull/9597#discussion_r502402326
##########
File path: airflow/api_connexion/endpoints/task_instance_endpoint.py
##########
@@ -14,39 +14,243 @@
# KIND, either express or implied. See the License for the
# specific language governing permissions and limitations
# under the License.
+from typing import Any, List, Optional, Tuple
+
+from flask import request
+from marshmallow import ValidationError
+from sqlalchemy import and_, func, or_
-# TODO(mik-laj): We have to implement it.
-# Do you want to help? Please look at: https://github.com/apache/airflow/issues/8132
from airflow.api_connexion import security
+from airflow.api_connexion.exceptions import NotFound, BadRequest
+from airflow.api_connexion.parameters import format_datetime, format_parameters
+from airflow.api_connexion.schemas.task_instance_schema import (
+ clear_task_instance_form,
+ TaskInstanceCollection,
+ task_instance_collection_schema,
+ task_instance_schema,
+ task_instance_batch_form,
+ task_instance_reference_collection_schema,
+ TaskInstanceReferenceCollection,
+)
+from airflow.configuration import conf
+from airflow.models.dagrun import DagRun as DR
+from airflow.models.taskinstance import clear_task_instances, TaskInstance as TI
+from airflow.models.dag import DagModel
+from airflow.models.dagbag import DagBag
+from airflow.models import SlaMiss
+from airflow.utils.state import State
+from airflow.utils.session import provide_session
@security.requires_authentication
-def get_task_instance():
- """
- Get a task instance
- """
- raise NotImplementedError("Not implemented yet.")
+@provide_session
+def get_task_instance(dag_id: str, dag_run_id: str, task_id: str, session=None):
+ """Get task instance"""
+ query = (
+ session.query(TI)
+ .filter(TI.dag_id == dag_id)
+ .join(DR, and_(TI.dag_id == DR.dag_id, TI.execution_date == DR.execution_date))
+ .filter(DR.run_id == dag_run_id)
+ .filter(TI.task_id == task_id)
+ .join(
+ SlaMiss,
+ and_(
+ SlaMiss.dag_id == TI.dag_id,
+ SlaMiss.execution_date == TI.execution_date,
+ SlaMiss.task_id == TI.task_id,
+ ),
+ isouter=True,
+ )
+ .add_entity(SlaMiss)
+ )
+ task_instance = query.one_or_none()
+ if task_instance is None:
+ raise NotFound("Task instance not found")
+
+ return task_instance_schema.dump(task_instance)
+
+
+def _apply_array_filter(query, key, values):
+ if values is not None:
+ query = query.filter(or_(*[key == v for v in values]))
+ return query
+
+
+def _apply_range_filter(query, key, value_range: Tuple[Any, Any]):
+ gte_value, lte_value = value_range
+ if gte_value is not None:
+ query = query.filter(key >= gte_value)
+ if lte_value is not None:
+ query = query.filter(key <= lte_value)
+ return query
+@format_parameters(
+ {
+ "execution_date_gte": format_datetime,
+ "execution_date_lte": format_datetime,
+ "start_date_gte": format_datetime,
+ "start_date_lte": format_datetime,
+ "end_date_gte": format_datetime,
+ "end_date_lte": format_datetime,
+ }
+)
@security.requires_authentication
-def get_task_instances():
+@provide_session
+def get_task_instances(
+ limit: int,
+ dag_id: Optional[str] = None,
+ dag_run_id: Optional[str] = None,
+ execution_date_gte: Optional[str] = None,
+ execution_date_lte: Optional[str] = None,
+ start_date_gte: Optional[str] = None,
+ start_date_lte: Optional[str] = None,
+ end_date_gte: Optional[str] = None,
+ end_date_lte: Optional[str] = None,
+ duration_gte: Optional[float] = None,
+ duration_lte: Optional[float] = None,
+ state: Optional[str] = None,
+ pool: Optional[List[str]] = None,
+ queue: Optional[List[str]] = None,
+ offset: Optional[int] = None,
+ session=None,
+): # pylint: disable=too-many-arguments
"""
- Get list of task instances of DAG.
+ Get list of task instances.
"""
- raise NotImplementedError("Not implemented yet.")
+ query = session.query(TI)
+
+ if dag_id != "~":
+ query = query.filter(TI.dag_id == dag_id)
+ if dag_run_id != "~":
+ query = query.join(DR, and_(TI.dag_id == DR.dag_id, TI.execution_date == DR.execution_date))
+ query = query.filter(DR.run_id == dag_run_id)
+ query = _apply_range_filter(
+ query,
+ key=DR.execution_date,
+ value_range=(execution_date_gte, execution_date_lte),
+ )
+ query = _apply_range_filter(query, key=TI.start_date, value_range=(start_date_gte, start_date_lte))
+ query = _apply_range_filter(query, key=TI.end_date, value_range=(end_date_gte, end_date_lte))
+ query = _apply_range_filter(query, key=TI.duration, value_range=(duration_gte, duration_lte))
+ query = _apply_array_filter(query, key=TI.state, values=state)
+ query = _apply_array_filter(query, key=TI.pool, values=pool)
+ query = _apply_array_filter(query, key=TI.queue, values=queue)
+
+ query = query.join(
+ SlaMiss,
+ and_(
+ SlaMiss.dag_id == TI.dag_id,
+ SlaMiss.task_id == TI.task_id,
+ SlaMiss.execution_date == TI.execution_date,
+ ),
+ isouter=True,
+ ).add_entity(SlaMiss)
+ task_instances = query.offset(offset).limit(limit).all()
+ total_entries = session.query(
+ func.count(func.distinct(TI.dag_id, TI.execution_date, TI.task_id))
+ ).scalar()
Review comment:
It seems to me that the count query doesn't need join, so we can simplify everything and improve SQL compatibility.
----------------------------------------------------------------
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 #9597: [WIP] Add read-only endpoints for task instances
Posted by GitBox <gi...@apache.org>.
kaxil commented on pull request #9597:
URL: https://github.com/apache/airflow/pull/9597#issuecomment-703748662
@OmairK @mik-laj Any updates here?
----------------------------------------------------------------
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] jhtimmins commented on a change in pull request #9597: [WIP] Add read-only endpoints for task instances
Posted by GitBox <gi...@apache.org>.
jhtimmins commented on a change in pull request #9597:
URL: https://github.com/apache/airflow/pull/9597#discussion_r451266925
##########
File path: airflow/api_connexion/endpoints/task_instance_endpoint.py
##########
@@ -14,23 +14,110 @@
# KIND, either express or implied. See the License for the
# specific language governing permissions and limitations
# under the License.
+from typing import Any, List, Optional, Tuple
-# TODO(mik-laj): We have to implement it.
-# Do you want to help? Please look at: https://github.com/apache/airflow/issues/8132
+from sqlalchemy import and_, func, or_
+from airflow.api_connexion.exceptions import NotFound
+from airflow.api_connexion.parameters import format_datetime, format_parameters
+from airflow.api_connexion.schemas.task_instance_schema import (
+ TaskInstanceCollection, task_instance_collection_schema, task_instance_schema,
+)
+from airflow.models.dagrun import DagRun as DR
+from airflow.models.taskinstance import TaskInstance as TI
+from airflow.utils.session import provide_session
+
+
+@provide_session
+def get_task_instance(dag_id: str, dag_run_id: str, task_id: str, session=None):
+ query = (
+ session.query(TI)
+ .filter(TI.dag_id == dag_id)
+ .join(DR, and_(TI.dag_id == DR.dag_id, TI.execution_date == DR.execution_date))
+ .filter(DR.run_id == dag_run_id)
+ .filter(TI.task_id == task_id)
+ )
+
+ task_instance = query.one_or_none()
+
+ if task_instance is None:
+ raise NotFound("Task instance not found")
+
+ return task_instance_schema.dump(task_instance)
-def get_task_instance():
- """
- Get a task instance
- """
- raise NotImplementedError("Not implemented yet.")
+def _apply_array_filter(query, key, values):
Review comment:
Mmmm actually nevermind. I think it makes sense.
----------------------------------------------------------------
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] github-actions[bot] commented on pull request #9597: Add endpoints for task instances
Posted by GitBox <gi...@apache.org>.
github-actions[bot] commented on pull request #9597:
URL: https://github.com/apache/airflow/pull/9597#issuecomment-706231622
[The Workflow run](https://github.com/apache/airflow/actions/runs/297696278) is cancelling this PR. It has some failed jobs matching ^Pylint$,^Static checks$,^Build docs$,^Spell check docs$,^Backport packages$,^Checks: Helm tests$,^Test OpenAPI*.
----------------------------------------------------------------
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 #9597: Add endpoints for task instances
Posted by GitBox <gi...@apache.org>.
kaxil commented on a change in pull request #9597:
URL: https://github.com/apache/airflow/pull/9597#discussion_r503852964
##########
File path: airflow/api_connexion/endpoints/task_instance_endpoint.py
##########
@@ -14,39 +14,234 @@
# KIND, either express or implied. See the License for the
# specific language governing permissions and limitations
# under the License.
+from typing import Any, List, Optional, Tuple
+
+from flask import request
+from marshmallow import ValidationError
+from sqlalchemy import and_, func, or_
-# TODO(mik-laj): We have to implement it.
-# Do you want to help? Please look at: https://github.com/apache/airflow/issues/8132
from airflow.api_connexion import security
+from airflow.api_connexion.exceptions import NotFound, BadRequest
+from airflow.api_connexion.parameters import format_datetime, format_parameters
+from airflow.api_connexion.schemas.task_instance_schema import (
+ clear_task_instance_form,
+ TaskInstanceCollection,
+ task_instance_collection_schema,
+ task_instance_schema,
+ task_instance_batch_form,
+ task_instance_reference_collection_schema,
+ TaskInstanceReferenceCollection,
+)
+from airflow.configuration import conf
+from airflow.models.dagrun import DagRun as DR
+from airflow.models.taskinstance import clear_task_instances, TaskInstance as TI
+from airflow.models.dag import DagModel
+from airflow.models.dagbag import DagBag
+from airflow.models import SlaMiss
+from airflow.utils.state import State
+from airflow.utils.session import provide_session
@security.requires_authentication
-def get_task_instance():
- """
- Get a task instance
- """
- raise NotImplementedError("Not implemented yet.")
+@provide_session
+def get_task_instance(dag_id: str, dag_run_id: str, task_id: str, session=None):
+ """Get task instance"""
+ query = (
+ session.query(TI)
+ .filter(TI.dag_id == dag_id)
+ .join(DR, and_(TI.dag_id == DR.dag_id, TI.execution_date == DR.execution_date))
+ .filter(DR.run_id == dag_run_id)
+ .filter(TI.task_id == task_id)
+ .join(
+ SlaMiss,
+ and_(
+ SlaMiss.dag_id == TI.dag_id,
+ SlaMiss.execution_date == TI.execution_date,
+ SlaMiss.task_id == TI.task_id,
+ ),
+ isouter=True,
+ )
+ .add_entity(SlaMiss)
+ )
+ task_instance = query.one_or_none()
+
+ if task_instance is None:
+ raise NotFound("Task instance not found")
+
+ return task_instance_schema.dump(task_instance)
+
+
+def _apply_array_filter(query, key, values):
+ if values is not None:
+ query = query.filter(or_(*[key == v for v in values]))
+ return query
+
+
+def _apply_range_filter(query, key, value_range: Tuple[Any, Any]):
+ gte_value, lte_value = value_range
+ if gte_value is not None:
+ query = query.filter(key >= gte_value)
+ if lte_value is not None:
+ query = query.filter(key <= lte_value)
+ return query
+@format_parameters(
+ {
+ "execution_date_gte": format_datetime,
+ "execution_date_lte": format_datetime,
+ "start_date_gte": format_datetime,
+ "start_date_lte": format_datetime,
+ "end_date_gte": format_datetime,
+ "end_date_lte": format_datetime,
+ }
+)
@security.requires_authentication
-def get_task_instances():
+@provide_session
+def get_task_instances(
+ limit: int,
+ dag_id: Optional[str] = None,
+ dag_run_id: Optional[str] = None,
+ execution_date_gte: Optional[str] = None,
+ execution_date_lte: Optional[str] = None,
+ start_date_gte: Optional[str] = None,
+ start_date_lte: Optional[str] = None,
+ end_date_gte: Optional[str] = None,
+ end_date_lte: Optional[str] = None,
+ duration_gte: Optional[float] = None,
+ duration_lte: Optional[float] = None,
+ state: Optional[str] = None,
+ pool: Optional[List[str]] = None,
+ queue: Optional[List[str]] = None,
+ offset: Optional[int] = None,
+ session=None,
+): # pylint: disable=too-many-arguments
"""
- Get list of task instances of DAG.
+ Get list of task instances.
"""
- raise NotImplementedError("Not implemented yet.")
+ query = session.query(TI)
+
+ if dag_id != "~":
+ query = query.filter(TI.dag_id == dag_id)
+ if dag_run_id != "~":
+ query = query.join(DR, and_(TI.dag_id == DR.dag_id, TI.execution_date == DR.execution_date))
+ query = query.filter(DR.run_id == dag_run_id)
+ query = _apply_range_filter(
+ query, key=DR.execution_date, value_range=(execution_date_gte, execution_date_lte),
+ )
+ query = _apply_range_filter(query, key=TI.start_date, value_range=(start_date_gte, start_date_lte))
+ query = _apply_range_filter(query, key=TI.end_date, value_range=(end_date_gte, end_date_lte))
+ query = _apply_range_filter(query, key=TI.duration, value_range=(duration_gte, duration_lte))
+ query = _apply_array_filter(query, key=TI.state, values=state)
+ query = _apply_array_filter(query, key=TI.pool, values=pool)
+ query = _apply_array_filter(query, key=TI.queue, values=queue)
+
+ query = query.join(
+ SlaMiss,
+ and_(
+ SlaMiss.dag_id == TI.dag_id,
+ SlaMiss.task_id == TI.task_id,
+ SlaMiss.execution_date == TI.execution_date,
+ ),
+ isouter=True,
+ ).add_entity(SlaMiss)
+ task_instances = query.offset(offset).limit(limit).all()
+ total_entries = session.query(
+ func.count(func.distinct(TI.dag_id, TI.execution_date, TI.task_id))
+ ).scalar()
+ return task_instance_collection_schema.dump(
+ TaskInstanceCollection(task_instances=task_instances, total_entries=total_entries)
+ )
@security.requires_authentication
-def get_task_instances_batch():
+@provide_session
+def get_task_instances_batch(session=None):
"""
Get list of task instances.
"""
- raise NotImplementedError("Not implemented yet.")
+ body = request.get_json()
+ try:
+ data = task_instance_batch_form.load(body)
+ except ValidationError as err:
+ raise BadRequest(detail=str(err.messages))
+ query = session.query(TI)
+
+ query = _apply_array_filter(query, key=TI.dag_id, values=data["dag_ids"])
+ query = _apply_range_filter(
+ query, key=TI.execution_date, value_range=(data["execution_date_gte"], data["execution_date_lte"]),
+ )
+ query = _apply_range_filter(
+ query, key=TI.start_date, value_range=(data["start_date_gte"], data["start_date_lte"]),
+ )
+ query = _apply_range_filter(
+ query, key=TI.end_date, value_range=(data["end_date_gte"], data["end_date_lte"])
+ )
+ query = _apply_range_filter(
+ query, key=TI.duration, value_range=(data["duration_gte"], data["duration_lte"])
+ )
+ query = _apply_array_filter(query, key=TI.state, values=data["state"])
+ query = _apply_array_filter(query, key=TI.pool, values=data["pool"])
+ query = _apply_array_filter(query, key=TI.queue, values=data["queue"])
+
+ query = query.join(
+ SlaMiss,
+ and_(
+ SlaMiss.dag_id == TI.dag_id,
+ SlaMiss.task_id == TI.task_id,
+ SlaMiss.execution_date == TI.execution_date,
+ ),
+ isouter=True,
+ ).add_entity(SlaMiss)
+ task_instances = query.all()
+ total_entries = session.query(
+ func.count(func.distinct(TI.dag_id, TI.execution_date, TI.task_id))
+ ).scalar()
+
+ return task_instance_collection_schema.dump(
+ TaskInstanceCollection(task_instances=task_instances, total_entries=total_entries)
+ )
@security.requires_authentication
-def post_clear_task_instances():
+@provide_session
+def post_clear_task_instances(dag_id: str, session=None):
Review comment:
this has been added in `post_set_task_instances_state` method below
----------------------------------------------------------------
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