Create task pipeline in DL-Python

[Ben-Hodgkiss]

We need to create tasks based on different logs produced by the dataset-resource pipeline and the collection pipeline. This is primarily to be used in async after collecting and transforming a resource. We should create this in digital-land-python incase we want to re-use it elsewhere.

Technical Approach

• in the pipeline subpackage create a task.py module and add the TaskPipeline class in there. it should have a run method which can be used to run the pipeline. arguments can either be fed in via the method arguments or pass in a pydantic dataclass Config into the TaskPipeline classwhich contains the relevant information. This is simalar to the pipelines found in the pyspark-jobs repo.
• it should create the task according to the agreed task schema. this is in the spec however we can ignore start-date and end-date
• it needs to create tasks from:
  • log (Collection Log) - it should create a row in the task data if the collection was not successful
  • issue (Issue log) - group issues by issue_type resource and field dataset, focus on issues with error severity and external responsibility for now but allow this to change in the future. one task row should represent an issue type for a particular field.
• Use polars to run transformations on the input csvs to generate a single task output csv. The polars functions can live within the task.py module for now. When we need to use then elsewhere we can create a separate python subpackage for polars transformation functions.

expected json for detail column for log tasks:

{
  "status":500
  "exception":""
}

expected json for detail for issue tasks:

{
  "issue_type":"invalid geom"
  "count":10
  "field":"geometry"
}

Acceptance criteria:

• create code to do the above, create an example output that can be shared and signed off by the team
• merge in code to digital-land-python - no need yet to use this code in async that's the next step

[Ben-Hodgkiss]

[Ben-Hodgkiss]

Link to task dataset: https://github.com/digital-land/specification/blob/main/content/dataset/task.md?plain=1

[tombrooks248]

Some notes from what Owen said in Sprint Planning - transcript:

So essentially what's happening right now is you've got the check service processing a single resource, looking at all the different types of logs that are generated and coming together.

If you go to the UI for the check tool, a list of tasks for someone to look at. What we found is that's really useful for people to be able to summarise: "here's the things that you need to fix" without having one row for every single value that might be wrong in their data.
So the idea is to have the task table look at several different source data sets or tables.

For example the collection log, the issue log, the column field log. I should then combine all of these into a list of tasks that need to be sorted. It should kind of acts as an index.

For example, lets say, there's 400 rows in the issue log that a user should look at and fix. But it summarises that in an easy to see one place for people to have a look at.

Currently the task list exists purely in the UI of the check tool right now, and it also exists in the LPA dashboard, and in both instances, it's having to be calculated on the fly in the provider's app. It's not being calculated before then, which can slow everything Down.

What we need to do is have the same task list that is produced in both places.

The plan for how to make this change is that we move from the current situation where we create a task list on the UI side in the check tool. Instead it will create it in the async back end; we want that to create a list of tasks, put that into the async request API so that it can be accessed by the front end. Then instead of the front end having to kind of calculate that more manually, it just goes to the task list and just displays it to the user.

Where to place the code:
We're going to build the actual code to do it in Digital Land Python, because Digital land Python is used by the async request backend. The async request backend is what is called by the submit application or the provide application. We should just create a file here, put it in this. It's likely that we will want to refactor this in the future, but this a good place to start. Just create this task, CSV or JSON or whatever we need to do so that it can be used in the async to upload that into the database and then accessed by the UI.

The reason for doing the async only first is that it's a very nice exploratory space for us to get this working. Then when it comes to doing it in the overnight pipeline, there are bigger questions at play about how we generate these tasks, because you're not just displaying tasks to someone in the LPL dashboard from a single file.

My suggestion is to shift to kind of using polars, create a pipeline class that uses polar functions to generate this stuff. As we can have essentially the first PR, we'll just put it into digital and Python with a this should work.