Coverage for /pythoncovmergedfiles/medio/medio/src/airflow/airflow/serialization/serializers/iceberg.py: 29%
Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1#
2# Licensed to the Apache Software Foundation (ASF) under one
3# or more contributor license agreements. See the NOTICE file
4# distributed with this work for additional information
5# regarding copyright ownership. The ASF licenses this file
6# to you under the Apache License, Version 2.0 (the
7# "License"); you may not use this file except in compliance
8# with the License. You may obtain a copy of the License at
9#
10# http://www.apache.org/licenses/LICENSE-2.0
11#
12# Unless required by applicable law or agreed to in writing,
13# software distributed under the License is distributed on an
14# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
15# KIND, either express or implied. See the License for the
16# specific language governing permissions and limitations
17# under the License.
18from __future__ import annotations
20from typing import TYPE_CHECKING
22from airflow.utils.module_loading import qualname
24serializers = ["pyiceberg.table.Table"]
25deserializers = serializers
26stringifiers = serializers
28if TYPE_CHECKING:
29 from airflow.serialization.serde import U
31__version__ = 1
34def serialize(o: object) -> tuple[U, str, int, bool]:
35 from pyiceberg.table import Table
37 if not isinstance(o, Table):
38 return "", "", 0, False
40 from airflow.models.crypto import get_fernet
42 # we encrypt the catalog information here until we have
43 # global catalog management in airflow and the properties
44 # can have sensitive information
45 fernet = get_fernet()
46 properties = {}
47 for k, v in o.catalog.properties.items():
48 properties[k] = fernet.encrypt(v.encode("utf-8")).decode("utf-8")
50 data = {
51 "identifier": o.identifier,
52 "catalog_properties": properties,
53 }
55 return data, qualname(o), __version__, True
58def deserialize(classname: str, version: int, data: dict):
59 from pyiceberg.catalog import load_catalog
60 from pyiceberg.table import Table
62 from airflow.models.crypto import get_fernet
64 if version > __version__:
65 raise TypeError("serialized version is newer than class version")
67 if classname == qualname(Table):
68 fernet = get_fernet()
69 properties = {}
70 for k, v in data["catalog_properties"].items():
71 properties[k] = fernet.decrypt(v.encode("utf-8")).decode("utf-8")
73 catalog = load_catalog(data["identifier"][0], **properties)
74 return catalog.load_table((data["identifier"][1], data["identifier"][2]))
76 raise TypeError(f"do not know how to deserialize {classname}")