You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "bbovenzi (via GitHub)" <gi...@apache.org> on 2026/05/21 17:39:05 UTC
[PR] Add asset and task state UI [airflow]
bbovenzi opened a new pull request, #67292:
URL: https://github.com/apache/airflow/pull/67292
Add CRUDs action for Asset and Task State
Asset State:
- Create a new tab navigation on an asset page to switch between events and asset state
Task State:
- Move xcoms and task state into a "Storage" tab with an xcom and task state sub-tabs. Xcoms url is preserved.
<img width="1864" height="858" alt="Screenshot 2026-05-20 at 6 04 12 PM" src="https://github.com/user-attachments/assets/70053c7a-cb55-4890-b9c4-eae2574b19e2" />
<img width="1869" height="860" alt="Screenshot 2026-05-20 at 6 04 35 PM" src="https://github.com/user-attachments/assets/a7577239-270c-48b0-8465-aa103cf32cf7" />
---
##### Was generative AI tooling used to co-author this PR?
<!--
If generative AI tooling has been used in the process of authoring this PR, please
change below checkbox to `[X]` followed by the name of the tool, uncomment the "Generated-by".
-->
- [ ] Yes (please specify the tool below)
<!--
Generated-by: [Tool Name] following [the guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#gen-ai-assisted-contributions)
-->
---
* Read the **[Pull Request Guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#pull-request-guidelines)** for more information. Note: commit author/co-author name and email in commits become permanently public when merged.
* For fundamental code changes, an Airflow Improvement Proposal ([AIP](https://cwiki.apache.org/confluence/display/AIRFLOW/Airflow+Improvement+Proposals)) is needed.
* When adding dependency, check compliance with the [ASF 3rd Party License Policy](https://www.apache.org/legal/resolved.html#category-x).
* For significant user-facing changes create newsfragment: `{pr_number}.significant.rst`, in [airflow-core/newsfragments](https://github.com/apache/airflow/tree/main/airflow-core/newsfragments). You can add this file in a follow-up commit after the PR is created so you know the PR number.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add asset and task state UI [airflow]
Posted by "amoghrajesh (via GitHub)" <gi...@apache.org>.
amoghrajesh commented on code in PR #67292:
URL: https://github.com/apache/airflow/pull/67292#discussion_r3286801876
##########
airflow-core/src/airflow/ui/src/pages/Asset/AssetStateTab.tsx:
##########
@@ -0,0 +1,127 @@
+/*!
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+import { Flex, Text } from "@chakra-ui/react";
+import type { ColumnDef } from "@tanstack/react-table";
+import { useTranslation } from "react-i18next";
+
+import type { AssetStateResponse } from "openapi/requests";
+import { DataTable } from "src/components/DataTable";
+import { useTableURLState } from "src/components/DataTable/useTableUrlState";
+import { ErrorAlert } from "src/components/ErrorAlert";
+import RenderedJsonField from "src/components/RenderedJsonField";
+import Time from "src/components/Time";
+import { TruncatedText } from "src/components/TruncatedText";
+import { useListAssetStates } from "src/queries/useAssetState";
+
+import AddAssetStateButton from "./AddAssetStateButton";
+import ClearAllAssetStateButton from "./ClearAllAssetStateButton";
+import DeleteAssetStateButton from "./DeleteAssetStateButton";
+import EditAssetStateButton from "./EditAssetStateButton";
+
+type AssetStateTabProps = {
+ readonly assetId: number;
+};
+
+type ColumnsProps = {
+ readonly assetId: number;
+ readonly translate: (key: string) => string;
+};
+
+const getColumns = ({ assetId, translate }: ColumnsProps): Array<ColumnDef<AssetStateResponse>> => [
+ {
+ accessorKey: "key",
+ cell: ({ row: { original } }) => <Text>{original.key}</Text>,
+ header: translate("assetState.columns.key"),
+ },
+ {
+ accessorKey: "value",
+ cell: ({ row: { original } }) => {
+ let parsed: unknown;
+
+ try {
+ parsed = JSON.parse(original.value);
+ } catch {
+ // not JSON — render as plain text
+ }
+ const isJsonObject = parsed !== null && parsed !== undefined && typeof parsed === "object";
+
+ return isJsonObject ? (
+ <RenderedJsonField collapsed content={parsed as object} enableClipboard={false} />
+ ) : (
+ <TruncatedText text={original.value} />
+ );
+ },
+ enableSorting: false,
+ header: translate("assetState.columns.value"),
+ },
+ {
+ accessorKey: "updated_at",
+ cell: ({ row: { original } }) => <Time datetime={original.updated_at} />,
+ header: translate("assetState.columns.updatedAt"),
+ },
+ {
+ accessorKey: "actions",
+ cell: ({ row: { original } }) => (
+ <Flex justifyContent="end">
+ <EditAssetStateButton assetId={assetId} stateKey={original.key} />
+ <DeleteAssetStateButton assetId={assetId} stateKey={original.key} />
+ </Flex>
+ ),
+ enableSorting: false,
+ header: "",
+ },
+];
Review Comment:
Asset states do not expire theoritically, so is it worth adding a column for all rows saying "Expiry" - "Never"?
--
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.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add asset and task state UI [airflow]
Posted by "bbovenzi (via GitHub)" <gi...@apache.org>.
bbovenzi commented on code in PR #67292:
URL: https://github.com/apache/airflow/pull/67292#discussion_r3290686260
##########
airflow-core/src/airflow/ui/src/pages/Asset/AssetStateTab.tsx:
##########
@@ -0,0 +1,127 @@
+/*!
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+import { Flex, Text } from "@chakra-ui/react";
+import type { ColumnDef } from "@tanstack/react-table";
+import { useTranslation } from "react-i18next";
+
+import type { AssetStateResponse } from "openapi/requests";
+import { DataTable } from "src/components/DataTable";
+import { useTableURLState } from "src/components/DataTable/useTableUrlState";
+import { ErrorAlert } from "src/components/ErrorAlert";
+import RenderedJsonField from "src/components/RenderedJsonField";
+import Time from "src/components/Time";
+import { TruncatedText } from "src/components/TruncatedText";
+import { useListAssetStates } from "src/queries/useAssetState";
+
+import AddAssetStateButton from "./AddAssetStateButton";
+import ClearAllAssetStateButton from "./ClearAllAssetStateButton";
+import DeleteAssetStateButton from "./DeleteAssetStateButton";
+import EditAssetStateButton from "./EditAssetStateButton";
+
+type AssetStateTabProps = {
+ readonly assetId: number;
+};
+
+type ColumnsProps = {
+ readonly assetId: number;
+ readonly translate: (key: string) => string;
+};
+
+const getColumns = ({ assetId, translate }: ColumnsProps): Array<ColumnDef<AssetStateResponse>> => [
+ {
+ accessorKey: "key",
+ cell: ({ row: { original } }) => <Text>{original.key}</Text>,
+ header: translate("assetState.columns.key"),
+ },
+ {
+ accessorKey: "value",
+ cell: ({ row: { original } }) => {
+ let parsed: unknown;
+
+ try {
+ parsed = JSON.parse(original.value);
+ } catch {
+ // not JSON — render as plain text
+ }
+ const isJsonObject = parsed !== null && parsed !== undefined && typeof parsed === "object";
+
+ return isJsonObject ? (
+ <RenderedJsonField collapsed content={parsed as object} enableClipboard={false} />
+ ) : (
+ <TruncatedText text={original.value} />
+ );
+ },
+ enableSorting: false,
+ header: translate("assetState.columns.value"),
+ },
+ {
+ accessorKey: "updated_at",
+ cell: ({ row: { original } }) => <Time datetime={original.updated_at} />,
+ header: translate("assetState.columns.updatedAt"),
+ },
+ {
+ accessorKey: "actions",
+ cell: ({ row: { original } }) => (
+ <Flex justifyContent="end">
+ <EditAssetStateButton assetId={assetId} stateKey={original.key} />
+ <DeleteAssetStateButton assetId={assetId} stateKey={original.key} />
+ </Flex>
+ ),
+ enableSorting: false,
+ header: "",
+ },
+];
Review Comment:
I don't quite get this. Why would we add a column that always says expires never?
--
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.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add asset and task state UI [airflow]
Posted by "amoghrajesh (via GitHub)" <gi...@apache.org>.
amoghrajesh commented on PR #67292:
URL: https://github.com/apache/airflow/pull/67292#issuecomment-4541297876
This one will also serve as a good enhancement here: https://github.com/apache/airflow/pull/67530
--
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.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add asset and task store UI [airflow]
Posted by "amoghrajesh (via GitHub)" <gi...@apache.org>.
amoghrajesh commented on PR #67292:
URL: https://github.com/apache/airflow/pull/67292#issuecomment-4656939155
Amazing, thanks brent!
--
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.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add asset and task state UI [airflow]
Posted by "amoghrajesh (via GitHub)" <gi...@apache.org>.
amoghrajesh commented on code in PR #67292:
URL: https://github.com/apache/airflow/pull/67292#discussion_r3286900745
##########
airflow-core/src/airflow/ui/src/pages/Storage/TaskStatePage.tsx:
##########
@@ -0,0 +1,30 @@
+/*!
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+import { useParams } from "react-router-dom";
+
+import { TaskStateTab } from "./TaskStateTab";
+
+export const TaskStatePage = () => {
+ const { dagId = "~", mapIndex = "-1", runId = "~", taskId = "~" } = useParams();
+ const parsedMapIndex = mapIndex === "-1" || mapIndex === "~" ? -1 : parseInt(mapIndex, 10);
Review Comment:
When no `mapIndex` is in the URL, `useParams()` returns `undefined`, not "-1". non-mapped tasks could get the wrong value.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add asset and task state UI [airflow]
Posted by "amoghrajesh (via GitHub)" <gi...@apache.org>.
amoghrajesh commented on code in PR #67292:
URL: https://github.com/apache/airflow/pull/67292#discussion_r3341085818
##########
airflow-core/src/airflow/ui/src/pages/Asset/AssetStateTab.tsx:
##########
@@ -0,0 +1,127 @@
+/*!
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+import { Flex, Text } from "@chakra-ui/react";
+import type { ColumnDef } from "@tanstack/react-table";
+import { useTranslation } from "react-i18next";
+
+import type { AssetStateResponse } from "openapi/requests";
+import { DataTable } from "src/components/DataTable";
+import { useTableURLState } from "src/components/DataTable/useTableUrlState";
+import { ErrorAlert } from "src/components/ErrorAlert";
+import RenderedJsonField from "src/components/RenderedJsonField";
+import Time from "src/components/Time";
+import { TruncatedText } from "src/components/TruncatedText";
+import { useListAssetStates } from "src/queries/useAssetState";
+
+import AddAssetStateButton from "./AddAssetStateButton";
+import ClearAllAssetStateButton from "./ClearAllAssetStateButton";
+import DeleteAssetStateButton from "./DeleteAssetStateButton";
+import EditAssetStateButton from "./EditAssetStateButton";
+
+type AssetStateTabProps = {
+ readonly assetId: number;
+};
+
+type ColumnsProps = {
+ readonly assetId: number;
+ readonly translate: (key: string) => string;
+};
+
+const getColumns = ({ assetId, translate }: ColumnsProps): Array<ColumnDef<AssetStateResponse>> => [
+ {
+ accessorKey: "key",
+ cell: ({ row: { original } }) => <Text>{original.key}</Text>,
+ header: translate("assetState.columns.key"),
+ },
+ {
+ accessorKey: "value",
+ cell: ({ row: { original } }) => {
+ let parsed: unknown;
+
+ try {
+ parsed = JSON.parse(original.value);
+ } catch {
+ // not JSON — render as plain text
+ }
+ const isJsonObject = parsed !== null && parsed !== undefined && typeof parsed === "object";
+
+ return isJsonObject ? (
+ <RenderedJsonField collapsed content={parsed as object} enableClipboard={false} />
+ ) : (
+ <TruncatedText text={original.value} />
+ );
+ },
+ enableSorting: false,
+ header: translate("assetState.columns.value"),
+ },
+ {
+ accessorKey: "updated_at",
+ cell: ({ row: { original } }) => <Time datetime={original.updated_at} />,
+ header: translate("assetState.columns.updatedAt"),
+ },
+ {
+ accessorKey: "actions",
+ cell: ({ row: { original } }) => (
+ <Flex justifyContent="end">
+ <EditAssetStateButton assetId={assetId} stateKey={original.key} />
+ <DeleteAssetStateButton assetId={assetId} stateKey={original.key} />
+ </Flex>
+ ),
+ enableSorting: false,
+ header: "",
+ },
+];
Review Comment:
NVM no need to do 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.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add asset and task state UI [airflow]
Posted by "amoghrajesh (via GitHub)" <gi...@apache.org>.
amoghrajesh commented on code in PR #67292:
URL: https://github.com/apache/airflow/pull/67292#discussion_r3296167584
##########
airflow-core/src/airflow/ui/src/pages/Asset/AssetStateTab.tsx:
##########
@@ -0,0 +1,127 @@
+/*!
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+import { Flex, Text } from "@chakra-ui/react";
+import type { ColumnDef } from "@tanstack/react-table";
+import { useTranslation } from "react-i18next";
+
+import type { AssetStateResponse } from "openapi/requests";
+import { DataTable } from "src/components/DataTable";
+import { useTableURLState } from "src/components/DataTable/useTableUrlState";
+import { ErrorAlert } from "src/components/ErrorAlert";
+import RenderedJsonField from "src/components/RenderedJsonField";
+import Time from "src/components/Time";
+import { TruncatedText } from "src/components/TruncatedText";
+import { useListAssetStates } from "src/queries/useAssetState";
+
+import AddAssetStateButton from "./AddAssetStateButton";
+import ClearAllAssetStateButton from "./ClearAllAssetStateButton";
+import DeleteAssetStateButton from "./DeleteAssetStateButton";
+import EditAssetStateButton from "./EditAssetStateButton";
+
+type AssetStateTabProps = {
+ readonly assetId: number;
+};
+
+type ColumnsProps = {
+ readonly assetId: number;
+ readonly translate: (key: string) => string;
+};
+
+const getColumns = ({ assetId, translate }: ColumnsProps): Array<ColumnDef<AssetStateResponse>> => [
+ {
+ accessorKey: "key",
+ cell: ({ row: { original } }) => <Text>{original.key}</Text>,
+ header: translate("assetState.columns.key"),
+ },
+ {
+ accessorKey: "value",
+ cell: ({ row: { original } }) => {
+ let parsed: unknown;
+
+ try {
+ parsed = JSON.parse(original.value);
+ } catch {
+ // not JSON — render as plain text
+ }
+ const isJsonObject = parsed !== null && parsed !== undefined && typeof parsed === "object";
+
+ return isJsonObject ? (
+ <RenderedJsonField collapsed content={parsed as object} enableClipboard={false} />
+ ) : (
+ <TruncatedText text={original.value} />
+ );
+ },
+ enableSorting: false,
+ header: translate("assetState.columns.value"),
+ },
+ {
+ accessorKey: "updated_at",
+ cell: ({ row: { original } }) => <Time datetime={original.updated_at} />,
+ header: translate("assetState.columns.updatedAt"),
+ },
+ {
+ accessorKey: "actions",
+ cell: ({ row: { original } }) => (
+ <Flex justifyContent="end">
+ <EditAssetStateButton assetId={assetId} stateKey={original.key} />
+ <DeleteAssetStateButton assetId={assetId} stateKey={original.key} />
+ </Flex>
+ ),
+ enableSorting: false,
+ header: "",
+ },
+];
Review Comment:
Just thought it would be _better_ to add such a thing or a note at the top for people who aren't aware that asset states do not expire. There is no real need to do 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.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add asset and task state UI [airflow]
Posted by "amoghrajesh (via GitHub)" <gi...@apache.org>.
amoghrajesh commented on PR #67292:
URL: https://github.com/apache/airflow/pull/67292#issuecomment-4516717972
One other thing I noticed about adding a task / asset state is, UI allows adding things like incomplete jsons, something like:
```bash
curl --location --request PUT 'http://localhost:28080/api/v2/dags/my_dag/dagRuns/manual__2026-05-22T07:59:31.188183+00:00/taskInstances/t1/states/job_id' \
--header 'Content-Type: application/json' \
--header 'Authorization: ••••••' \
--data '{
"value": "incomplete
}'
```
The API responds with 422 error but UI is sending it as:
```bash
curl 'http://localhost:28080/api/v2/dags/my_dag/dagRuns/manual__2026-05-22T07:59:31.188183+00:00/taskInstances/t1/states/abcd?map_index=-1' \
-X 'PUT' \
--data-raw '{"value":"{\"abcd\": \"a}"}'
```
I think that the UI should validate the form field before constructing the JSON body to send
--
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.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add asset and task store UI [airflow]
Posted by "amoghrajesh (via GitHub)" <gi...@apache.org>.
amoghrajesh commented on PR #67292:
URL: https://github.com/apache/airflow/pull/67292#issuecomment-4645430861
Taking it for a spin again!
--
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.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add asset and task state UI [airflow]
Posted by "amoghrajesh (via GitHub)" <gi...@apache.org>.
amoghrajesh commented on PR #67292:
URL: https://github.com/apache/airflow/pull/67292#issuecomment-4517710687
I found another bug related to the core API where editing a task state field overwrote the `expiry` to `NEVER` and also we didn't provide an option for users to set expiry for a task state when creating a new one. For the task state Storage tab, here's how the UI should call the API:
Adding a new task state:
<img width="1665" height="728" alt="image" src="https://github.com/user-attachments/assets/30461934-50c8-496c-a7b4-e3f171b6592b" />
So the modal for adding a task state ^ will need a new field for `expiry_date` - I imagine a datetime picker along with something that can serve three expiry options:
1. "default" pre-selected
2. Maybe a radio button for "never expire"
3. Datetime picker for selecting a datetime for expiry
Once a value is picked, call the `PUT /states/{key}` with:
- `{"value": "...", "expires_at": "default"}` — apply server default retention
- `{"value": "...", "expires_at": null}` — never expire
`{"value": "...", "expires_at": "2026-06-01T00:00:00Z"}` — specific datetime from the date picker
Editing an existing task state:
Edit existing key (value only) now can call the `PATCH /states/{key}` with `{"value": "..."}`. Expiry is always preserved, no expiry field needed.
This is being fixed in https://github.com/apache/airflow/pull/67319
--
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.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add asset and task store UI [airflow]
Posted by "amoghrajesh (via GitHub)" <gi...@apache.org>.
amoghrajesh commented on PR #67292:
URL: https://github.com/apache/airflow/pull/67292#issuecomment-4645633357
> Thanks for the awesome work, @bbovenzi!
>
> Looking really nice, I left some comments on the PRs for issues after testing these many things, you have my dags but the last one for mapped is here:
>
> ```python
> from __future__ import annotations
>
> import json
> import random
> from datetime import datetime, timezone
>
> from airflow.sdk import DAG, task
>
> TABLES = ["orders", "customers", "products"]
>
> with DAG(
> dag_id="example_task_state_mapped",
> schedule=None,
> start_date=datetime(2026, 1, 1),
> catchup=False,
> tags=["example", "aip-103", "task-state", "mapped"],
> doc_md=__doc__,
> ) as dag:
>
> @task
> def get_tables() -> list[str]:
> """Return the list of tables to process."""
> return TABLES
>
> @task
> def process_table(table: str, **context) -> dict:
> """Process one table — each mapped instance gets its own task state."""
> ts = context["task_state"]
> map_index = context["task_instance"].map_index
>
> row_count = random.randint(100, 10000)
> result = {
> "table": table,
> "map_index": map_index,
> "row_count": row_count,
> "processed_at": datetime.now(tz=timezone.utc).isoformat(timespec="seconds"),
> }
>
> ts.set("table", table)
> ts.set("status", "complete")
> ts.set("row_count", str(row_count))
> ts.set("result", json.dumps(result))
>
> print(f"[map_index={map_index}] Processed {table}: {row_count} rows")
> return result
>
> tables = get_tables()
> process_table.expand(table=tables)
> ```
>
> ## Task State — Spark DAG
> * [x] All keys visible after a completed run (`job_id`, `submitted_at`, `status`, `poll_result`, `completed_at`)
> * [x] `poll_result` JSON is pretty-printed
> * [x] `job_id` shows **Never** in Expires At, other keys show a date
> * [x] After retry-reattach: same `job_id` persists, `status` updates to `complete`
> * [x] Delete a single key — row gone, others intact
> * [x] Edit a key — new value shows immediately
> * [x] Clear all — table goes empty
>
> ## Asset State — Watermark DAG
> * [x] First run: `watermark`, `total_runs=1`, `last_run_summary` appear on asset detail page
> * [x] Subsequent runs: `total_runs` increments, `watermark` advances, `prev_watermark` matches previous run
> * [x] Consumer DAG fires automatically after each producer run
> * [x] Clear asset state then re-trigger: `total_runs=1`, `prev_watermark=null`
>
> ## Mapped Tasks — Mapped DAG (`example_task_state_mapped`)
> * [x] Trigger DAG — 3 mapped instances run (map_index 0, 1, 2 for orders/customers/products)
> * [x] Each mapped TI shows its own `table`, `row_count`, `result` in Storage tab — no bleed between instances
> * [x] Switching between map_index 0/1/2 in the UI shows different state values
> * [x] Clear single instance (`map_index=0`) — only that instance's state is gone, others intact
> * [x] Clear all (`all_map_indices=true`) — state wiped across all 3 instances
I am retesting similar scenarios but with the example dags for task store and asset store committed to repo.
1. The section: ## Task State — Spark DAG still looks great! And love the validation here.
<img width="865" height="903" alt="image" src="https://github.com/user-attachments/assets/4af8c142-a3cf-4e12-a007-c80e7a3d886e" />
2. Same with: `Asset State — Watermark DAG` section, looks great.
Validation on UI is great too.
<img width="989" height="818" alt="image" src="https://github.com/user-attachments/assets/b6fa1c78-7af3-4c68-82dd-161512542b09" />
3. Using this dag for mapped
```python
from __future__ import annotations
import random
from datetime import datetime, timezone
from airflow.sdk import DAG, task
TABLES = ["orders", "customers", "products"]
with DAG(
dag_id="example_task_store_mapped",
schedule=None,
start_date=datetime(2026, 1, 1),
catchup=False,
tags=["example", "aip-103", "task-state", "mapped"],
doc_md=__doc__,
) as dag:
@task
def get_tables() -> list[str]:
"""Return the list of tables to process."""
return TABLES
@task
def process_table(table: str, **context) -> dict:
"""Process one table — each mapped instance gets its own task state."""
ts = context["task_store"]
map_index = context["task_instance"].map_index
row_count = random.randint(100, 10000)
result = {
"table": table,
"map_index": map_index,
"row_count": row_count,
"processed_at": datetime.now(tz=timezone.utc).isoformat(timespec="seconds"),
}
ts.set("table", table)
ts.set("status", "complete")
ts.set("row_count", row_count)
ts.set("result", result)
print(f"[map_index={map_index}] Processed {table}: {row_count} rows")
return result
tables = get_tables()
process_table.expand(table=tables)
```
Looks great too, tried, get, set, clear, clear all etc.
Now for custom backends, using this backend:
```python
# 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.
"""
File-based state backend for testing the ExternalState envelope.
Workers write state values as JSON files under /tmp/airflow_state/ and store
the file path as the external reference. The DB therefore holds:
{"__type": "ExternalState", "__var": "/tmp/airflow_state/ti_<id>/job_id.json"}
instead of the raw value, which lets you verify the envelope behaviour end-to-end.
Configure in airflow.cfg (or via env var) before starting a worker:
[workers]
state_backend = dev.file_state_backend.FileStateBackend
The server-side abstract methods (get/set/delete/clear and their async variants)
raise NotImplementedError — this backend is purely a worker-side serialization hook.
"""
from __future__ import annotations
import json
from pathlib import Path
from typing import TYPE_CHECKING
from airflow.sdk.state import BaseStoreBackend
if TYPE_CHECKING:
from datetime import datetime
from pydantic import JsonValue
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.orm import Session
from airflow_shared.state import StoreScope
BASE_DIR = Path("/tmp/airflow_state")
class FileStateBackend(BaseStoreBackend):
"""Stores task/asset state values as local JSON files; returns the path as the ref."""
def serialize_task_store_to_ref(self, *, value: JsonValue, key: str, ti_id: str) -> str:
path = BASE_DIR / f"ti_{ti_id}" / f"{key}.json"
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(json.dumps(value))
return str(path)
def deserialize_task_store_from_ref(self, stored: str) -> JsonValue:
return json.loads(Path(stored).read_text())
def serialize_asset_state_to_ref(self, *, value: JsonValue, key: str, asset_ref: str) -> str:
safe = asset_ref.replace("/", "_").replace(":", "")
path = BASE_DIR / "assets" / safe / f"{key}.json"
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(json.dumps(value))
return str(path)
def deserialize_asset_state_from_ref(self, stored: str) -> JsonValue:
return json.loads(Path(stored).read_text())
def get(self, scope: StoreScope, key: str, *, session: Session | None = None) -> str | None:
raise NotImplementedError(
"FileStateBackend is a worker-side backend; server uses MetastoreStateBackend"
)
def set(
self,
scope: StoreScope,
key: str,
value: str,
*,
expires_at: datetime | None = None,
session: Session | None = None,
) -> None:
raise NotImplementedError
def delete(self, scope: StoreScope, key: str, *, session: Session | None = None) -> None:
raise NotImplementedError
def clear(
self, scope: StoreScope, *, all_map_indices: bool = False, session: Session | None = None
) -> None:
raise NotImplementedError
async def aget(self, scope: StoreScope, key: str, *, session: AsyncSession | None = None) -> str | None:
raise NotImplementedError
async def aset(
self,
scope: StoreScope,
key: str,
value: str,
*,
expires_at: datetime | None = None,
session: AsyncSession | None = None,
) -> None:
raise NotImplementedError
async def adelete(self, scope: StoreScope, key: str, *, session: AsyncSession | None = None) -> None:
raise NotImplementedError
async def aclear(
self, scope: StoreScope, *, all_map_indices: bool = False, session: AsyncSession | None = None
) -> None:
raise NotImplementedError
```
And tried out whether the custom ref envelope shows up, and it looks fine:
<img width="1345" height="831" alt="image" src="https://github.com/user-attachments/assets/df3b64ad-5bef-4f3d-8ec3-c1f1e920be54" />
--
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.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add asset and task store UI [airflow]
Posted by "pierrejeambrun (via GitHub)" <gi...@apache.org>.
pierrejeambrun commented on code in PR #67292:
URL: https://github.com/apache/airflow/pull/67292#discussion_r3374334639
##########
airflow-core/src/airflow/ui/src/pages/TaskStore/DeleteTaskStoreButton.tsx:
##########
@@ -0,0 +1,82 @@
+/*!
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+import { useDisclosure } from "@chakra-ui/react";
+import { useQueryClient } from "@tanstack/react-query";
+import { useTranslation } from "react-i18next";
+import { FiTrash2 } from "react-icons/fi";
+
+import { useTaskStoreServiceDeleteTaskStore, useTaskStoreServiceListTaskStoreKey } from "openapi/queries";
+import DeleteDialog from "src/components/DeleteDialog";
+import { IconButton, toaster } from "src/components/ui";
+import { createErrorToaster } from "src/utils";
+
+type Props = {
+ readonly dagId: string;
+ readonly mapIndex: number;
+ readonly runId: string;
+ readonly storeKey: string;
+ readonly taskId: string;
+};
+
+export const DeleteTaskStoreButton = ({ dagId, mapIndex, runId, storeKey, taskId }: Props) => {
+ const { t: translate } = useTranslation(["dag", "common"]);
+ const { onClose, onOpen, open } = useDisclosure();
+ const queryClient = useQueryClient();
+
+ const { isPending, mutate } = useTaskStoreServiceDeleteTaskStore({
+ onError: (error) => {
+ createErrorToaster(
+ error,
+ { params: { resourceName: translate("taskStore.title") }, titleKey: "common:toaster.delete.error" },
+ translate,
+ );
+ },
+ onSuccess: async () => {
+ await queryClient.invalidateQueries({ queryKey: [useTaskStoreServiceListTaskStoreKey] });
Review Comment:
We probably need to also invalidate the `Get*StoreKey` in all 6 mutation hook. (That could be the cause of the issue mentioned by amogh -> stale value in the modal after successfull update)
##########
airflow-core/src/airflow/ui/src/pages/TaskStore/DeleteTaskStoreButton.tsx:
##########
@@ -0,0 +1,82 @@
+/*!
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+import { useDisclosure } from "@chakra-ui/react";
+import { useQueryClient } from "@tanstack/react-query";
+import { useTranslation } from "react-i18next";
+import { FiTrash2 } from "react-icons/fi";
+
+import { useTaskStoreServiceDeleteTaskStore, useTaskStoreServiceListTaskStoreKey } from "openapi/queries";
+import DeleteDialog from "src/components/DeleteDialog";
+import { IconButton, toaster } from "src/components/ui";
+import { createErrorToaster } from "src/utils";
+
+type Props = {
+ readonly dagId: string;
+ readonly mapIndex: number;
+ readonly runId: string;
+ readonly storeKey: string;
+ readonly taskId: string;
+};
+
+export const DeleteTaskStoreButton = ({ dagId, mapIndex, runId, storeKey, taskId }: Props) => {
+ const { t: translate } = useTranslation(["dag", "common"]);
+ const { onClose, onOpen, open } = useDisclosure();
+ const queryClient = useQueryClient();
+
+ const { isPending, mutate } = useTaskStoreServiceDeleteTaskStore({
+ onError: (error) => {
+ createErrorToaster(
+ error,
+ { params: { resourceName: translate("taskStore.title") }, titleKey: "common:toaster.delete.error" },
+ translate,
+ );
+ },
+ onSuccess: async () => {
+ await queryClient.invalidateQueries({ queryKey: [useTaskStoreServiceListTaskStoreKey] });
Review Comment:
nit: Also most of the handlers here are the same (toasts, invalidation, etc...) maybe a common `useStoreMutation` factorizing all that could be easier than having all the code duplicated. (Easier to miss / desync mutation / toasts etc..)
--
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.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add asset and task store UI [airflow]
Posted by "bbovenzi (via GitHub)" <gi...@apache.org>.
bbovenzi merged PR #67292:
URL: https://github.com/apache/airflow/pull/67292
--
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.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org