subhramit commented on code in PR #72310: URL: https://github.com/apache/airflow/pull/72310#discussion_r4090217537
########## shared/search/src/airflow_shared/search/response.py: ########## @@ -0,0 +1,202 @@ +# 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 __future__ import annotations + +from collections.abc import Iterator +from typing import Any + +__all__ = [ + "AttributeDict", + "AttributeList", + "Hit", + "HitMeta", + "SearchResponse", + "resolve_nested", +] + + +def _wrap(val): + if isinstance(val, dict): + return AttributeDict(val) + return val + + +def resolve_nested(hit: dict[Any, Any], parent_class=None) -> type[Hit]: + """ + Resolve nested hits from a search backend by iteratively navigating the ``_nested`` field. + + The result is used to fetch the appropriate document class to handle the hit. + + This method can be used with nested fields which are structured + as dictionaries with "field" and "_nested" keys. + """ + nested_path: list[str] = [] + nesting = hit["_nested"] + while nesting and "field" in nesting: + nested_path.append(nesting["field"]) + nesting = nesting.get("_nested") + nested_path_str = ".".join(nested_path) + + if hasattr(parent_class, "_index"): + nested_field = parent_class._index.resolve_field(nested_path_str) + if nested_field is not None: + return nested_field._doc_class + + return Hit + + +class AttributeList: + """Helper class to provide attribute like access to List objects.""" + + def __init__(self, _list): + if not isinstance(_list, list): + _list = list(_list) + self._l_ = _list + + def __getitem__(self, k): + """Retrieve an item or a slice from the list. If the item is a dictionary, it is wrapped in an AttributeDict.""" + val = self._l_[k] + if isinstance(val, slice): Review Comment: Not introduced in this PR, but wouldn't this always be False? I think this should be: ```suggestion if isinstance(k, slice): ``` otherwise the slice will return a plain list instead. ########## providers/opensearch/src/airflow/providers/opensearch/log/os_json_formatter.py: ########## @@ -14,36 +14,12 @@ # KIND, either express or implied. See the License for the # specific language governing permissions and limitations # under the License. + from __future__ import annotations -from airflow.providers.opensearch.version_compat import AIRFLOW_V_3_3_PLUS +from airflow.providers.opensearch._shared.search.formatter import ISO8601Formatter Review Comment: I think (as per my IDE), both providers' `AIRFLOW_V_3_3_PLUS` in `version_compat.py` will now be unused? SHould that then be cleaned up? ########## shared/search/src/airflow_shared/search/response.py: ########## @@ -0,0 +1,202 @@ +# 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 __future__ import annotations + +from collections.abc import Iterator +from typing import Any + +__all__ = [ + "AttributeDict", + "AttributeList", + "Hit", + "HitMeta", + "SearchResponse", + "resolve_nested", +] + + +def _wrap(val): + if isinstance(val, dict): + return AttributeDict(val) + return val + + +def resolve_nested(hit: dict[Any, Any], parent_class=None) -> type[Hit]: + """ + Resolve nested hits from a search backend by iteratively navigating the ``_nested`` field. + + The result is used to fetch the appropriate document class to handle the hit. + + This method can be used with nested fields which are structured + as dictionaries with "field" and "_nested" keys. + """ + nested_path: list[str] = [] + nesting = hit["_nested"] + while nesting and "field" in nesting: + nested_path.append(nesting["field"]) + nesting = nesting.get("_nested") + nested_path_str = ".".join(nested_path) + + if hasattr(parent_class, "_index"): + nested_field = parent_class._index.resolve_field(nested_path_str) + if nested_field is not None: + return nested_field._doc_class + + return Hit + + +class AttributeList: + """Helper class to provide attribute like access to List objects.""" + + def __init__(self, _list): + if not isinstance(_list, list): + _list = list(_list) + self._l_ = _list + + def __getitem__(self, k): + """Retrieve an item or a slice from the list. If the item is a dictionary, it is wrapped in an AttributeDict.""" + val = self._l_[k] + if isinstance(val, slice): Review Comment: What, however, is introduced in this PR is `test_slice_access_returns_plain_list` which makes this bug the expected behavior. That would also need fixing. ########## dev/breeze/src/airflow_breeze/templates/pyproject_TEMPLATE.toml.jinja2: ########## @@ -179,6 +179,9 @@ include = [ "docs", "src/airflow/providers/{{ PROVIDER_PATH }}", "tests", +{% if SHARED_DISTRIBUTIONS %} + "provider.yaml", +{% endif %} Review Comment: Ah I see. The flit providers (including elasticsearch and opensearch) have `provider.yaml` in the sdist, but that is never the case for hatchling providers. Then can we use one rule for all hatchling providers? Either include `provider.yaml` always, or remove it for elasticsearch and opensearch too (if nothing reads it from the sdist). ########## dev/breeze/src/airflow_breeze/templates/pyproject_TEMPLATE.toml.jinja2: ########## @@ -179,6 +179,9 @@ include = [ "docs", "src/airflow/providers/{{ PROVIDER_PATH }}", "tests", +{% if SHARED_DISTRIBUTIONS %} + "provider.yaml", +{% endif %} Review Comment: Why is `provider.yaml` tied to `SHARED_DISTRIBUTIONS`? It does not seem to be related. ########## shared/search/src/airflow_shared/search/response.py: ########## @@ -0,0 +1,202 @@ +# 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 __future__ import annotations + +from collections.abc import Iterator +from typing import Any + +__all__ = [ + "AttributeDict", + "AttributeList", + "Hit", + "HitMeta", + "SearchResponse", + "resolve_nested", +] + + +def _wrap(val): + if isinstance(val, dict): + return AttributeDict(val) + return val + + +def resolve_nested(hit: dict[Any, Any], parent_class=None) -> type[Hit]: + """ + Resolve nested hits from a search backend by iteratively navigating the ``_nested`` field. + + The result is used to fetch the appropriate document class to handle the hit. + + This method can be used with nested fields which are structured + as dictionaries with "field" and "_nested" keys. + """ + nested_path: list[str] = [] + nesting = hit["_nested"] + while nesting and "field" in nesting: + nested_path.append(nesting["field"]) + nesting = nesting.get("_nested") + nested_path_str = ".".join(nested_path) + + if hasattr(parent_class, "_index"): + nested_field = parent_class._index.resolve_field(nested_path_str) + if nested_field is not None: + return nested_field._doc_class + + return Hit + + +class AttributeList: + """Helper class to provide attribute like access to List objects.""" + + def __init__(self, _list): + if not isinstance(_list, list): + _list = list(_list) + self._l_ = _list + + def __getitem__(self, k): + """Retrieve an item or a slice from the list. If the item is a dictionary, it is wrapped in an AttributeDict.""" + val = self._l_[k] + if isinstance(val, slice): + return AttributeList(val) + return _wrap(val) + + def __iter__(self): + """Provide an iterator for the list or the dictionary.""" + return (_wrap(i) for i in self._l_) + + def __bool__(self): + """Check if the list is non-empty.""" + return bool(self._l_) + + +class AttributeDict: + """Helper class to provide attribute like access to Dictionary objects.""" + + def __init__(self, d): + super().__setattr__("_d_", d) + + def __getattr__(self, attr_name): + """Retrieve an item as an attribute from the dictionary.""" + try: + return self.__getitem__(attr_name) + except KeyError: + raise AttributeError(f"{self.__class__.__name__!r} object has no attribute {attr_name!r}") + + def __getitem__(self, key): + """Retrieve an item using a key from the dictionary.""" + return _wrap(self._d_[key]) + + def to_dict(self): + return self._d_ + + +class Hit(AttributeDict): + """ + The Hit class is used to manage and access elements in a document. + + It inherits from the AttributeDict class and provides + attribute-like access to its elements, similar to a dictionary. + """ + + def __init__(self, document): + data = {} + if "_source" in document: + data = document["_source"] + if "fields" in document: + data.update(document["fields"]) Review Comment: This is also not introduced by the PR, but this changes the caller's dict. Perhaps a good opportunity to fix. ```suggestion def __init__(self, document): data = dict(document.get("_source", {})) if "fields" in document: data.update(document["fields"]) ``` Please also add a test for 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: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
