I'm not sure this should live inside API endpoint code. Fetching a task instance is pretty generic functionality. If we don't already support code that does this, we should probably add it to TaskInstance.py or similar files.

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

Ah sorry for being unclear. I was suggesting get_task_instance be moved to a different file outside the API.

)Can be a follow up PR (and likely candidate would be to TaskIstance.get_by_run_id())

This filter seems somewhat ambiguous. I think the name could better express the purpose.

I do not know if I understand correctly. Can you say more?

Mmmm actually nevermind. I think it makes sense.

This function accepts 16 parameters. If we really need to pass in this many configuration details, is there a data structure that would better encapsulate system state than so many standalone params?

We use connexion, which fills these parameters based on the API specification.
https://connexion.readthedocs.io/en/latest/request.html#automatic-parameter-handling

Looks like the preceding two lines are duplicates

Good point. I will remove it

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

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. 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

if you need it, we can start using relationships.

I'll try to look at it tomorrow.

Cool, in the meantime I will prepare this solution

@mik-laj I have dealt with the n +1 problem, but I am unsure if this query is the best way to go forward.

@mik-laj is there a better approach that I should work on or is this fine?

LGTM.

@mik-laj Passing multiple columns to distinct works perfectly with postgres but raises error in case of sqlite how should we handle this case.

It seems to me that the count query doesn't need join, so we can simplify everything and improve SQL compatibility.

if we want to do an outerjoin instead of isouter=True

Can we change single to double quotes here too for consistency

Fixed. :-)