pierrejeambrun commented on code in PR #51682:
URL: https://github.com/apache/airflow/pull/51682#discussion_r2263049532
##########
airflow-core/src/airflow/api_fastapi/core_api/routes/ui/dependencies.py:
##########
@@ -41,12 +42,27 @@
),
dependencies=[Depends(requires_access_dag("GET",
DagAccessEntity.DEPENDENCIES))],
)
-def get_dependencies(session: SessionDep, node_id: str | None = None) ->
BaseGraphResponse:
- """Dependencies graph."""
+def get_dependencies(
+ node_id: str | None = None,
+ node_ids: list[str] = Query(None, description="List of node ids"),
+) -> BaseGraphResponse:
+ """Dependencies graph. Supports a single node_id or multiple node_ids as
exploded query parameters (node_ids=foo&node_ids=bar)."""
Review Comment:
```suggestion
"""Dependencies graph. Supports a single node_id or multiple node_ids
```
##########
airflow-core/src/airflow/api_fastapi/core_api/routes/ui/dependencies.py:
##########
@@ -41,12 +42,27 @@
),
dependencies=[Depends(requires_access_dag("GET",
DagAccessEntity.DEPENDENCIES))],
)
-def get_dependencies(session: SessionDep, node_id: str | None = None) ->
BaseGraphResponse:
- """Dependencies graph."""
+def get_dependencies(
+ node_id: str | None = None,
Review Comment:
```suggestion
```
We should probably only use `node_ids` even for single node to simplify the
API. That's private API so it doens't have to be backward compatible.
##########
airflow-core/src/airflow/ui/src/vite-env.d.ts:
##########
@@ -19,6 +19,7 @@
*/
/// <reference types="vite/client" />
+/// <reference types="vite/types/importMeta.d.ts" />
Review Comment:
I don't think it's needed anymore. That must have been fixed in other PRs.
##########
airflow-core/src/airflow/api_fastapi/core_api/routes/public/assets.py:
##########
@@ -125,7 +127,7 @@ def depends(cls, only_active: bool = True) ->
OnlyActiveFilter:
def get_assets(
limit: QueryLimit,
offset: QueryOffset,
- name_pattern: QueryAssetNamePatternSearch,
+ asset_pattern_filter: QueryAssetPatternFilter,
Review Comment:
We shouldn't replace like this, but add the `group_pattern_filter` next to
it, so we can chose what to filter on.
##########
airflow-core/src/airflow/api_fastapi/core_api/routes/public/assets.py:
##########
@@ -206,6 +208,88 @@ def get_assets(
)
+@assets_router.get(
+ "/assets/groups",
+ responses=create_openapi_http_exception_doc([status.HTTP_404_NOT_FOUND]),
+ dependencies=[
+ Depends(requires_access_asset(method="GET")),
+ ],
+)
+def get_asset_groups(
+ limit: QueryLimit,
+ offset: QueryOffset,
+ only_active: Annotated[OnlyActiveFilter,
Depends(OnlyActiveFilter.depends)],
+ session: SessionDep,
+ order_by: Annotated[str, Query(alias="order_by")] = "group",
+ group: str | None = Query(default=None, description="Filter by grupo"),
+) -> AssetGroupCollectionResponse:
+ """Get asset groups."""
+ offset_value = offset.value if offset.value is not None else 0
+
+ group_query = (
+ select(AssetModel.group, func.count(AssetModel.id))
+ .where(
+ AssetModel.active.has() if only_active.value else True,
+ *([AssetModel.group == group] if group else []),
+ )
+ .group_by(AssetModel.group)
+ )
+ group_rows = session.execute(group_query).all()
+ limit_value = limit.value if limit.value is not None else len(group_rows)
+ # Ordene em Python
+ if order_by == "count":
+ group_rows.sort(key=lambda x: (x[1], x[0]))
+ elif order_by == "-count":
+ group_rows.sort(key=lambda x: (-x[1], x[0]))
+ elif order_by == "-group":
+ group_rows.sort(key=lambda x: x[0], reverse=True)
+ else:
+ group_rows.sort(key=lambda x: x[0])
+ # Aplique paginação em Python
+ paginated_rows = group_rows[offset_value : offset_value + limit_value]
+ groups = [row[0] for row in paginated_rows]
+ group_counts = {row[0]: row[1] for row in paginated_rows}
+
+ # Query all assets for these groups
+ assets_query = (
+ select(AssetModel)
+ .where(AssetModel.group.in_(groups))
+ .options(
+ subqueryload(AssetModel.scheduled_dags),
+ subqueryload(AssetModel.producing_tasks),
+ subqueryload(AssetModel.consuming_tasks),
+ )
+ )
+ assets = session.scalars(assets_query).all()
Review Comment:
This will query all assets from the db. Can't happen I think
##########
airflow-core/src/airflow/ui/src/pages/Asset/AssetGroupSidebar.tsx:
##########
@@ -0,0 +1,97 @@
+/*!
+ * 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 { Box, Heading, Text, HStack, VStack, Badge, Button } from
"@chakra-ui/react";
+import { useTranslation } from "react-i18next";
+
+import type { AssetResponse } from "openapi/requests/types.gen";
+
+type AssetGroupSidebarProps = {
+ assets: Array<AssetResponse>;
+ groupName: string;
+};
+
+export const AssetGroupSidebar: React.FC<AssetGroupSidebarProps> = ({ assets,
groupName }) => {
+ const { t: translate } = useTranslation(["assets", "common"]);
+
+ const producingTasks = assets.reduce((acc, asset) => acc +
asset.producing_tasks.length, 0);
+ const consumingDags = assets.reduce((acc, asset) => acc +
asset.consuming_tasks.length, 0);
+
+ // Collect all last asset events from all assets
+ const events = assets
+ .map((asset) => asset.last_asset_event)
+ .filter((ev): ev is NonNullable<typeof ev> => Boolean(ev));
+
+ return (
+ <VStack align="stretch" gap={4}>
+ <Box bg="chakra-subtle-bg" borderColor="chakra-border-color"
borderRadius="md" borderWidth={1} p={4}>
+ <HStack gap={2} mb={2}>
+ <Badge aria-label={translate("group", { ns: "assets" })}
colorScheme="gray" fontSize="lg">
+ {translate("databaseIcon", { ns: "assets" })}
+ </Badge>
+ <Heading size="md">{groupName}</Heading>
+ </HStack>
+ <Text color="chakra-fg" fontSize="sm" mb={2}>
+ <b>{translate("group", { ns: "assets" })}</b>
+ <br />
+ {groupName}
+ </Text>
+ <HStack gap={4} mb={2}>
+ <Box>
+ <Text color="chakra-fg" fontSize="xs">
+ {translate("producingTasks", { ns: "assets" })}
+ </Text>
+ <Button disabled size="sm" variant="outline">
+ {producingTasks} {translate("task_one", { count: producingTasks,
ns: "common" })}
+ </Button>
+ </Box>
+ <Box>
+ <Text color="chakra-fg" fontSize="xs">
+ {translate("consumingDags", { ns: "assets" })}
+ </Text>
+ <Button disabled size="sm" variant="outline">
+ {consumingDags} {translate("dag_one", { count: consumingDags,
ns: "common" })}
+ </Button>
+ </Box>
+ </HStack>
+ </Box>
+ <Box bg="chakra-subtle-bg" borderColor="chakra-border-color"
borderRadius="md" borderWidth={1} p={4}>
+ <HStack mb={2}>
+ <Badge colorScheme="blue" />
+ <Heading size="sm">{translate("lastAssetEvent", { ns: "assets"
})}</Heading>
+ </HStack>
+ {events.length === 0 ? (
+ <Text color="chakra-fg-subtle" fontSize="sm">
+ {translate("noAssetEvents", { ns: "dashboard" })}
+ </Text>
+ ) : (
+ events.map((ev) => (
+ <Box bg="chakra-subtle-bg-hover" borderRadius="md" key={ev.id}
mb={2} p={2}>
+ <Text color="chakra-fg-subtle" fontSize="xs">
+ {ev.timestamp}
+ </Text>
Review Comment:
We probably shouldn't display timestamp as text, check other location of the
app for `Time` component.
##########
airflow-core/src/airflow/ui/src/pages/AssetsList/AssetsList.tsx:
##########
@@ -92,64 +105,80 @@ const createColumns = (translate: (key: string) =>
string): Array<ColumnDef<Asse
},
];
-const NAME_PATTERN_PARAM = SearchParamsKeys.NAME_PATTERN;
-
export const AssetsList = () => {
const { t: translate } = useTranslation(["assets", "common"]);
const [searchParams, setSearchParams] = useSearchParams();
- const [namePattern, setNamePattern] =
useState(searchParams.get(NAME_PATTERN_PARAM) ?? undefined);
+ const [searchValue, setSearchValue] = useState(
+ searchParams.get(SearchParamsKeys.NAME_PATTERN) ??
+ searchParams.get(SearchParamsKeys.GROUP_PATTERN) ??
+ undefined,
+ );
const { setTableURLState, tableURLState } = useTableURLState();
const { pagination, sorting } = tableURLState;
const [sort] = sorting;
const orderBy = sort ? `${sort.desc ? "-" : ""}${sort.id}` : undefined;
+ const searchMatchMode = "any"; // "any" para OR, "all" para AND
Review Comment:
comment not in english.
##########
airflow-core/src/airflow/ui/src/queries/useMultiDependencyGraph.ts:
##########
@@ -0,0 +1,53 @@
+/*!
+ * 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 { useQueryClient, type UseQueryOptions } from "@tanstack/react-query";
+
+import {
+ useDependenciesServiceGetDependencies,
+ UseDependenciesServiceGetDependenciesKeyFn,
+} from "openapi/queries";
+import type { BaseGraphResponse } from "openapi/requests/types.gen";
+
+/**
+ * Hook to fetch the dependency graph for a list of asset ids.
+ * Uses the backend endpoint that accepts node_ids as an exploded array
(node_ids=foo&node_ids=bar).
+ * Caches the result by ids.
+ */
+export const useMultiDependencyGraph = (
+ assetIds: Array<string>,
+ options?: Omit<UseQueryOptions<BaseGraphResponse, unknown>, "queryFn" |
"queryKey">,
+) => {
+ const queryClient = useQueryClient();
+
+ // Always send nodeIds as an array of asset:... strings
+ const nodeIdsParam = assetIds.map((id) => (id.startsWith("asset:") ? id :
`asset:${id}`));
+
+ const query = useDependenciesServiceGetDependencies({ nodeIds: nodeIdsParam
}, undefined, options);
+
+ if (query.data) {
+ const key = UseDependenciesServiceGetDependenciesKeyFn({ nodeIds:
nodeIdsParam });
+ const queryData = queryClient.getQueryData(key);
+
+ if (!Boolean(queryData)) {
+ queryClient.setQueryData(key, query.data);
+ }
+ }
+
+ return query;
+};
Review Comment:
That should `useDependencyGraph.ts` we don't need another hook doing the
exact same thing.
##########
airflow-core/src/airflow/ui/src/pages/AssetsGroupedList/AssetsGroupedList.tsx:
##########
@@ -0,0 +1,282 @@
+/*!
+ * 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 { Box, Link, IconButton, VStack, Flex, Heading, Text, Input } from
"@chakra-ui/react";
+import { useState, useMemo } from "react";
+import { useTranslation } from "react-i18next";
+import { FaChevronDown, FaChevronRight, FaChevronUp } from "react-icons/fa";
+import { Link as RouterLink } from "react-router-dom";
+
+import { useAssetServiceGetAssetGroups } from "openapi/queries";
+import type { AssetResponse } from "openapi/requests/types.gen";
+import { DataTable } from "src/components/DataTable";
+import { useTableURLState } from "src/components/DataTable/useTableUrlState";
+import { ErrorAlert } from "src/components/ErrorAlert";
+import Time from "src/components/Time";
+
+import { CreateAssetEvent } from "../Asset/CreateAssetEvent";
+import { DependencyPopover } from "../AssetsList/DependencyPopover";
+
+const NameCell = ({ original }: { readonly original: AssetResponse }) => (
+ <Link asChild color="fg.info" fontWeight="bold">
+ <RouterLink to={`/assets/${original.id}`}>{original.name}</RouterLink>
+ </Link>
+);
+
+const LastAssetEventCell = ({ original }: { readonly original: AssetResponse
}) => {
+ const assetEvent = original.last_asset_event;
+ const timestamp = assetEvent?.timestamp;
+
+ if (timestamp === undefined || timestamp === null) {
+ return undefined;
+ }
+
+ return <Time datetime={timestamp} />;
+};
+
+const GroupCell = ({ original }: { readonly original: AssetResponse }) => {
+ const { group } = original;
+
+ if (!group) {
+ return undefined;
+ }
+
+ return (
+ <Link asChild color="fg.info" fontWeight="bold">
+ <RouterLink to={`/assets/group/${group}`}>{group}</RouterLink>
+ </Link>
+ );
+};
+
+const ConsumingDagsCell = ({ original }: { readonly original: AssetResponse })
=>
+ original.consuming_tasks.length ? (
+ <DependencyPopover dependencies={original.consuming_tasks} type="Dag" />
+ ) : undefined;
+
+const ProducingTasksCell = ({ original }: { readonly original: AssetResponse
}) =>
+ original.producing_tasks.length ? (
+ <DependencyPopover dependencies={original.producing_tasks} type="Task" />
+ ) : undefined;
+
+const TriggerCell = ({ original }: { readonly original: AssetResponse }) => (
+ <CreateAssetEvent asset={original} withText={false} />
+);
+
+const truncate = (str: string, max = 32) => (str.length > max ?
`${str.slice(0, max - 3)}...` : str);
+
+const nameCellRenderer = ({ row: { original } }: { row: { original:
AssetResponse } }) => (
+ <NameCell original={original} />
+);
+const lastAssetEventCellRenderer = ({ row: { original } }: { row: { original:
AssetResponse } }) => (
+ <LastAssetEventCell original={original} />
+);
+const groupCellRenderer = ({ row: { original } }: { row: { original:
AssetResponse } }) => (
+ <GroupCell original={original} />
+);
+const consumingDagsCellRenderer = ({ row: { original } }: { row: { original:
AssetResponse } }) => (
+ <ConsumingDagsCell original={original} />
+);
+const producingTasksCellRenderer = ({ row: { original } }: { row: { original:
AssetResponse } }) => (
+ <ProducingTasksCell original={original} />
+);
+const triggerCellRenderer = ({ row: { original } }: { row: { original:
AssetResponse } }) => (
+ <TriggerCell original={original} />
+);
+
+export const AssetsGroupedList = () => {
+ const { t: translate } = useTranslation(["assets", "common"]);
+ const { setTableURLState, tableURLState } = useTableURLState();
+ const { pagination, sorting } = tableURLState;
+ const [sort] = sorting;
+ const orderBy = sort ? `${sort.desc ? "-" : ""}${sort.id}` : undefined;
+
+ // Use o novo hook para buscar grupos paginados
Review Comment:
Comment not in english.
##########
airflow-core/src/airflow/ui/src/pages/AssetsGroupedList/AssetsGroupedList.tsx:
##########
@@ -0,0 +1,282 @@
+/*!
+ * 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 { Box, Link, IconButton, VStack, Flex, Heading, Text, Input } from
"@chakra-ui/react";
+import { useState, useMemo } from "react";
+import { useTranslation } from "react-i18next";
+import { FaChevronDown, FaChevronRight, FaChevronUp } from "react-icons/fa";
+import { Link as RouterLink } from "react-router-dom";
+
+import { useAssetServiceGetAssetGroups } from "openapi/queries";
+import type { AssetResponse } from "openapi/requests/types.gen";
+import { DataTable } from "src/components/DataTable";
+import { useTableURLState } from "src/components/DataTable/useTableUrlState";
+import { ErrorAlert } from "src/components/ErrorAlert";
+import Time from "src/components/Time";
+
+import { CreateAssetEvent } from "../Asset/CreateAssetEvent";
+import { DependencyPopover } from "../AssetsList/DependencyPopover";
+
+const NameCell = ({ original }: { readonly original: AssetResponse }) => (
+ <Link asChild color="fg.info" fontWeight="bold">
+ <RouterLink to={`/assets/${original.id}`}>{original.name}</RouterLink>
+ </Link>
+);
+
+const LastAssetEventCell = ({ original }: { readonly original: AssetResponse
}) => {
+ const assetEvent = original.last_asset_event;
+ const timestamp = assetEvent?.timestamp;
+
+ if (timestamp === undefined || timestamp === null) {
+ return undefined;
+ }
+
+ return <Time datetime={timestamp} />;
+};
+
+const GroupCell = ({ original }: { readonly original: AssetResponse }) => {
+ const { group } = original;
+
+ if (!group) {
+ return undefined;
+ }
+
+ return (
+ <Link asChild color="fg.info" fontWeight="bold">
+ <RouterLink to={`/assets/group/${group}`}>{group}</RouterLink>
+ </Link>
+ );
+};
+
+const ConsumingDagsCell = ({ original }: { readonly original: AssetResponse })
=>
+ original.consuming_tasks.length ? (
+ <DependencyPopover dependencies={original.consuming_tasks} type="Dag" />
+ ) : undefined;
+
+const ProducingTasksCell = ({ original }: { readonly original: AssetResponse
}) =>
+ original.producing_tasks.length ? (
+ <DependencyPopover dependencies={original.producing_tasks} type="Task" />
+ ) : undefined;
+
+const TriggerCell = ({ original }: { readonly original: AssetResponse }) => (
+ <CreateAssetEvent asset={original} withText={false} />
+);
+
+const truncate = (str: string, max = 32) => (str.length > max ?
`${str.slice(0, max - 3)}...` : str);
+
+const nameCellRenderer = ({ row: { original } }: { row: { original:
AssetResponse } }) => (
+ <NameCell original={original} />
+);
+const lastAssetEventCellRenderer = ({ row: { original } }: { row: { original:
AssetResponse } }) => (
+ <LastAssetEventCell original={original} />
+);
+const groupCellRenderer = ({ row: { original } }: { row: { original:
AssetResponse } }) => (
+ <GroupCell original={original} />
+);
+const consumingDagsCellRenderer = ({ row: { original } }: { row: { original:
AssetResponse } }) => (
+ <ConsumingDagsCell original={original} />
+);
+const producingTasksCellRenderer = ({ row: { original } }: { row: { original:
AssetResponse } }) => (
+ <ProducingTasksCell original={original} />
+);
+const triggerCellRenderer = ({ row: { original } }: { row: { original:
AssetResponse } }) => (
+ <TriggerCell original={original} />
+);
+
Review Comment:
Similarly to other columns definition, inline this directly in the columns
definition.
##########
airflow-core/src/airflow/ui/src/queries/useMultiDependencyGraph.ts:
##########
@@ -0,0 +1,53 @@
+/*!
+ * 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 { useQueryClient, type UseQueryOptions } from "@tanstack/react-query";
+
+import {
+ useDependenciesServiceGetDependencies,
+ UseDependenciesServiceGetDependenciesKeyFn,
+} from "openapi/queries";
+import type { BaseGraphResponse } from "openapi/requests/types.gen";
+
+/**
+ * Hook to fetch the dependency graph for a list of asset ids.
+ * Uses the backend endpoint that accepts node_ids as an exploded array
(node_ids=foo&node_ids=bar).
+ * Caches the result by ids.
+ */
Review Comment:
```suggestion
```
##########
airflow-core/src/airflow/ui/src/queries/useMultiDependencyGraph.ts:
##########
@@ -0,0 +1,53 @@
+/*!
+ * 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 { useQueryClient, type UseQueryOptions } from "@tanstack/react-query";
+
+import {
+ useDependenciesServiceGetDependencies,
+ UseDependenciesServiceGetDependenciesKeyFn,
+} from "openapi/queries";
+import type { BaseGraphResponse } from "openapi/requests/types.gen";
+
+/**
+ * Hook to fetch the dependency graph for a list of asset ids.
+ * Uses the backend endpoint that accepts node_ids as an exploded array
(node_ids=foo&node_ids=bar).
+ * Caches the result by ids.
+ */
+export const useMultiDependencyGraph = (
+ assetIds: Array<string>,
+ options?: Omit<UseQueryOptions<BaseGraphResponse, unknown>, "queryFn" |
"queryKey">,
+) => {
+ const queryClient = useQueryClient();
+
+ // Always send nodeIds as an array of asset:... strings
+ const nodeIdsParam = assetIds.map((id) => (id.startsWith("asset:") ? id :
`asset:${id}`));
+
+ const query = useDependenciesServiceGetDependencies({ nodeIds: nodeIdsParam
}, undefined, options);
+
+ if (query.data) {
+ const key = UseDependenciesServiceGetDependenciesKeyFn({ nodeIds:
nodeIdsParam });
+ const queryData = queryClient.getQueryData(key);
+
+ if (!Boolean(queryData)) {
+ queryClient.setQueryData(key, query.data);
+ }
Review Comment:
This is done automatically by react query. Since you are setting the cache
for `nodeIdsParam` which is one of the query parameters. That is not needed.
##########
airflow-core/src/airflow/api_fastapi/core_api/routes/public/assets.py:
##########
@@ -206,6 +208,88 @@ def get_assets(
)
+@assets_router.get(
+ "/assets/groups",
+ responses=create_openapi_http_exception_doc([status.HTTP_404_NOT_FOUND]),
+ dependencies=[
+ Depends(requires_access_asset(method="GET")),
+ ],
+)
+def get_asset_groups(
+ limit: QueryLimit,
+ offset: QueryOffset,
+ only_active: Annotated[OnlyActiveFilter,
Depends(OnlyActiveFilter.depends)],
+ session: SessionDep,
+ order_by: Annotated[str, Query(alias="order_by")] = "group",
+ group: str | None = Query(default=None, description="Filter by grupo"),
+) -> AssetGroupCollectionResponse:
+ """Get asset groups."""
+ offset_value = offset.value if offset.value is not None else 0
+
+ group_query = (
+ select(AssetModel.group, func.count(AssetModel.id))
+ .where(
+ AssetModel.active.has() if only_active.value else True,
+ *([AssetModel.group == group] if group else []),
+ )
+ .group_by(AssetModel.group)
+ )
+ group_rows = session.execute(group_query).all()
+ limit_value = limit.value if limit.value is not None else len(group_rows)
+ # Ordene em Python
Review Comment:
Please english comments
##########
airflow-core/src/airflow/api_fastapi/core_api/routes/public/assets.py:
##########
@@ -206,6 +208,88 @@ def get_assets(
)
+@assets_router.get(
+ "/assets/groups",
+ responses=create_openapi_http_exception_doc([status.HTTP_404_NOT_FOUND]),
+ dependencies=[
+ Depends(requires_access_asset(method="GET")),
+ ],
+)
+def get_asset_groups(
+ limit: QueryLimit,
+ offset: QueryOffset,
+ only_active: Annotated[OnlyActiveFilter,
Depends(OnlyActiveFilter.depends)],
+ session: SessionDep,
+ order_by: Annotated[str, Query(alias="order_by")] = "group",
+ group: str | None = Query(default=None, description="Filter by grupo"),
+) -> AssetGroupCollectionResponse:
+ """Get asset groups."""
+ offset_value = offset.value if offset.value is not None else 0
+
+ group_query = (
+ select(AssetModel.group, func.count(AssetModel.id))
+ .where(
+ AssetModel.active.has() if only_active.value else True,
+ *([AssetModel.group == group] if group else []),
+ )
+ .group_by(AssetModel.group)
+ )
+ group_rows = session.execute(group_query).all()
+ limit_value = limit.value if limit.value is not None else len(group_rows)
+ # Ordene em Python
+ if order_by == "count":
+ group_rows.sort(key=lambda x: (x[1], x[0]))
+ elif order_by == "-count":
+ group_rows.sort(key=lambda x: (-x[1], x[0]))
+ elif order_by == "-group":
+ group_rows.sort(key=lambda x: x[0], reverse=True)
+ else:
+ group_rows.sort(key=lambda x: x[0])
+ # Aplique paginação em Python
Review Comment:
same
##########
airflow-core/src/airflow/api_fastapi/core_api/routes/public/assets.py:
##########
@@ -206,6 +208,88 @@ def get_assets(
)
+@assets_router.get(
+ "/assets/groups",
+ responses=create_openapi_http_exception_doc([status.HTTP_404_NOT_FOUND]),
+ dependencies=[
+ Depends(requires_access_asset(method="GET")),
+ ],
+)
+def get_asset_groups(
+ limit: QueryLimit,
+ offset: QueryOffset,
+ only_active: Annotated[OnlyActiveFilter,
Depends(OnlyActiveFilter.depends)],
+ session: SessionDep,
+ order_by: Annotated[str, Query(alias="order_by")] = "group",
+ group: str | None = Query(default=None, description="Filter by grupo"),
+) -> AssetGroupCollectionResponse:
+ """Get asset groups."""
+ offset_value = offset.value if offset.value is not None else 0
+
+ group_query = (
+ select(AssetModel.group, func.count(AssetModel.id))
+ .where(
+ AssetModel.active.has() if only_active.value else True,
+ *([AssetModel.group == group] if group else []),
+ )
+ .group_by(AssetModel.group)
+ )
+ group_rows = session.execute(group_query).all()
+ limit_value = limit.value if limit.value is not None else len(group_rows)
+ # Ordene em Python
+ if order_by == "count":
+ group_rows.sort(key=lambda x: (x[1], x[0]))
+ elif order_by == "-count":
+ group_rows.sort(key=lambda x: (-x[1], x[0]))
+ elif order_by == "-group":
+ group_rows.sort(key=lambda x: x[0], reverse=True)
+ else:
+ group_rows.sort(key=lambda x: x[0])
Review Comment:
What is that? Sorting should be on the db side, not on the application side.
##########
airflow-core/src/airflow/api_fastapi/common/parameters.py:
##########
@@ -665,7 +665,53 @@ def _transform_ti_states(states: list[str] | None) ->
list[TaskInstanceState | N
),
]
+
# Assets
+class _AssetPatternFilterModel(BaseModel):
+ """Asset Pattern Filter Model with a match mode parameter."""
+
+ name_pattern: str | None = None
+ group_pattern: str | None = None
+ search_match_mode: Literal["any", "all"] | None = None
Review Comment:
Maybe don't do advanced match mode directly cause we would need to do some
refactoring with `_TagsFilter` to have a BaseAdvance match mode filter to
extend from. Because there's a lot duplicated.
##########
airflow-core/src/airflow/api_fastapi/core_api/services/ui/dependencies.py:
##########
@@ -59,20 +66,34 @@ def extract_single_connected_component(
connected_components = extract_connected_components(adjacency_matrix)
- filtered_connected_components = [cc for cc in connected_components if
node_id in cc]
-
- if len(filtered_connected_components) != 1:
- raise ValueError(
- f"Unique connected component not found, got
{filtered_connected_components} for connected components of node {node_id},
expected only 1 connected component."
- )
-
- connected_component = filtered_connected_components[0]
-
- nodes = [node for node in nodes if node["id"] in connected_component]
- edges = [
+ if len(node_ids) == 1:
+ node_id = node_ids[0]
+ filtered_connected_components = [cc for cc in connected_components if
node_id in cc]
+ if len(filtered_connected_components) != 1:
+ raise ValueError(
+ f"Unique connected component not found, got
{filtered_connected_components} for connected components of node {node_id},
expected only 1 connected component."
+ )
+ connected_component = filtered_connected_components[0]
+ filtered_nodes = [node for node in nodes if node["id"] in
connected_component]
+ filtered_edges = [
+ edge
+ for edge in edges
+ if (edge["source_id"] in connected_component and edge["target_id"]
in connected_component)
+ ]
+ return {"nodes": filtered_nodes, "edges": filtered_edges}
+ # Multiple node_ids: merge all relevant components
+ relevant_components = [cc for cc in connected_components if any(nid in cc
for nid in node_ids)]
+ # NEW: Check that all node_ids are present in the merged set
+ if not relevant_components:
+ raise ValueError(f"No connected component found for node(s)
{node_ids}.")
+ merged_node_ids = set().union(*relevant_components)
+ missing = [nid for nid in node_ids if nid not in merged_node_ids]
+ if missing:
+ raise ValueError(f"No connected component found for node(s)
{missing}.")
+ filtered_nodes = [node for node in nodes if node["id"] in merged_node_ids]
+ filtered_edges = [
edge
for edge in edges
- if (edge["source_id"] in connected_component and edge["target_id"] in
connected_component)
+ if (edge["source_id"] in merged_node_ids and edge["target_id"] in
merged_node_ids)
Review Comment:
Needs refactoring, that's really hard to read .
##########
airflow-core/src/airflow/ui/src/pages/Asset/AssetGroupSidebar.tsx:
##########
@@ -0,0 +1,97 @@
+/*!
+ * 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 { Box, Heading, Text, HStack, VStack, Badge, Button } from
"@chakra-ui/react";
+import { useTranslation } from "react-i18next";
+
+import type { AssetResponse } from "openapi/requests/types.gen";
+
+type AssetGroupSidebarProps = {
+ assets: Array<AssetResponse>;
+ groupName: string;
+};
+
+export const AssetGroupSidebar: React.FC<AssetGroupSidebarProps> = ({ assets,
groupName }) => {
+ const { t: translate } = useTranslation(["assets", "common"]);
+
+ const producingTasks = assets.reduce((acc, asset) => acc +
asset.producing_tasks.length, 0);
+ const consumingDags = assets.reduce((acc, asset) => acc +
asset.consuming_tasks.length, 0);
+
+ // Collect all last asset events from all assets
+ const events = assets
+ .map((asset) => asset.last_asset_event)
+ .filter((ev): ev is NonNullable<typeof ev> => Boolean(ev));
+
+ return (
+ <VStack align="stretch" gap={4}>
+ <Box bg="chakra-subtle-bg" borderColor="chakra-border-color"
borderRadius="md" borderWidth={1} p={4}>
+ <HStack gap={2} mb={2}>
+ <Badge aria-label={translate("group", { ns: "assets" })}
colorScheme="gray" fontSize="lg">
+ {translate("databaseIcon", { ns: "assets" })}
+ </Badge>
+ <Heading size="md">{groupName}</Heading>
+ </HStack>
+ <Text color="chakra-fg" fontSize="sm" mb={2}>
+ <b>{translate("group", { ns: "assets" })}</b>
+ <br />
Review Comment:
use chakra.p, chakra.br, etc.
##########
airflow-core/src/airflow/api_fastapi/common/parameters.py:
##########
@@ -665,7 +665,53 @@ def _transform_ti_states(states: list[str] | None) ->
list[TaskInstanceState | N
),
]
+
# Assets
+class _AssetPatternFilterModel(BaseModel):
+ """Asset Pattern Filter Model with a match mode parameter."""
+
+ name_pattern: str | None = None
+ group_pattern: str | None = None
+ search_match_mode: Literal["any", "all"] | None = None
Review Comment:
Those should probably be two different params, not just one.
And you can define them easily with the search_param_factory since they are
not really doing more.
I don't think you need a custom class
```
QueryDagIdPatternSearch = Annotated[
_SearchParam, Depends(search_param_factory(DagModel.dag_id,
"dag_id_pattern"))
]
QueryDagDisplayNamePatternSearch = Annotated[
_SearchParam, Depends(search_param_factory(DagModel.dag_display_name,
"dag_display_name_pattern"))
]
```
##########
airflow-core/src/airflow/api_fastapi/core_api/routes/public/assets.py:
##########
@@ -206,6 +208,88 @@ def get_assets(
)
+@assets_router.get(
+ "/assets/groups",
+ responses=create_openapi_http_exception_doc([status.HTTP_404_NOT_FOUND]),
+ dependencies=[
+ Depends(requires_access_asset(method="GET")),
+ ],
+)
+def get_asset_groups(
+ limit: QueryLimit,
+ offset: QueryOffset,
+ only_active: Annotated[OnlyActiveFilter,
Depends(OnlyActiveFilter.depends)],
+ session: SessionDep,
+ order_by: Annotated[str, Query(alias="order_by")] = "group",
+ group: str | None = Query(default=None, description="Filter by grupo"),
+) -> AssetGroupCollectionResponse:
+ """Get asset groups."""
+ offset_value = offset.value if offset.value is not None else 0
+
+ group_query = (
+ select(AssetModel.group, func.count(AssetModel.id))
+ .where(
+ AssetModel.active.has() if only_active.value else True,
+ *([AssetModel.group == group] if group else []),
+ )
+ .group_by(AssetModel.group)
+ )
Review Comment:
This will query all assets from the db grouped by asset group, but that
could be thousands of them, that' not scalable.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
