migrate from davidkhala.data.format to davidkhala.data.integration

Author: davidkhala

.github/workflows/data.yaml

@@ -10,8 +10,6 @@ jobs:
           tests: format/tests
           test-entry-point: pytest
           working-directory: format
-        env:
-          PRIVATE_KEY: ${{secrets.PRIVATE_KEY}}
   frame:
     runs-on: ubuntu-latest
     steps:
@@ -30,4 +28,15 @@ jobs:
       - run: export DOCKER_BUILDKIT=1
       - run: docker build -t dash:latest visualization/app/dash
       - run: docker run --name $container -d dash:latest
-      - run: curl https://raw.githubusercontent.com/davidkhala/containers/refs/heads/main/cli/health.sh | bash -s wait-until-healthy $container
+      - run: curl https://raw.githubusercontent.com/davidkhala/containers/refs/heads/main/cli/health.sh | bash -s wait-until-healthy $container
+  integration:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@main
+      - uses: davidkhala/uv-buildpack@main
+        with:
+          tests: integration/tests
+          test-entry-point: pytest
+          working-directory: integration
+        env:
+          PRIVATE_KEY: ${{secrets.PRIVATE_KEY}}

…vidkhala/data/format/parquet/__init__.py …/davidkhala/data/format/arrow/parquet.pyformat/davidkhala/data/format/parquet/__init__.py renamed to format/davidkhala/data/format/arrow/parquet.py format/davidkhala/data/format/parquet/__init__.py renamed to format/davidkhala/data/format/arrow/parquet.py

@@ -17,4 +17,4 @@ def read_batch(self) -> Table:
         return self.file.read()
 
     def read_stream(self) -> Iterator[RecordBatch]:
-        return self.file.iter_batches()
+        return self.file.iter_batches()

format/pyproject.toml

@@ -3,7 +3,7 @@ name = "davidkhala.data.format"
 version = "0.0.5"
 description = ""
 authors = [{ name = "David Liu", email = "david-khala@hotmail.com" }]
-requires-python = "~=3.10"
+requires-python = ">=3.10"
 readme = "README.md"
 
 [project.optional-dependencies]

format/tests/arrow_test.py

@@ -1,11 +1,9 @@
-import os
 import unittest
 
-from pyarrow import input_stream
-
 from davidkhala.data.format.arrow.gcp import GCS
-from davidkhala.data.format.arrow.local_fs import LocalFS
-from davidkhala.data.format.parquet import Parquet
+from davidkhala.data.format.arrow.parquet import Parquet
+from pyarrow import input_stream
+from pyarrow.parquet import ColumnSchema, ParquetSchema, ParquetLogicalType
 
 
 class Samples(unittest.TestCase):
@@ -31,51 +29,25 @@ def test_GCS(self):
         with fs.open_input_stream(file_list[0]) as f:
             self.assertEqual(f.read(64), b'GROUP = FILE_HEADER\n  LANDSAT_SCENE_ID = "LC80010032013082LGN03"')
 
-    def test_parquet2arrow(self):
-        parquet = Parquet('fixtures/gcp-data-davidkhala.dbt_davidkhala.country_codes.parquet')
-        arrow_batch_path = 'artifacts/gcp-data-davidkhala.dbt_davidkhala.country_codes.batch.arrow'
-
-        fs = LocalFS()
-        fs.overwrite = True
-        fs.write_batch(arrow_batch_path, parquet.read_batch())
-        arrow_stream_path = 'artifacts/gcp-data-davidkhala.dbt_davidkhala.country_codes.stream.arrow'
-        fs.write_stream(arrow_stream_path, parquet.read_stream())
-
-
-class GCSTests(unittest.TestCase):
-    """
-    tests on private bucket
-    """
-    bucket = "davidkhala-data"
-
-    def test_ADC(self):
-        """
-        based on GCP ADC
-        """
-        for file in self.gcs.ls(self.bucket):
-            print(file.path)
-
-    @property
-    def gcs(self):
-        private_key = os.environ.get("PRIVATE_KEY")
-        if private_key:
-            return GCS.from_service_account({
-                'client_email': 'data-integration@gcp-data-davidkhala.iam.gserviceaccount.com',
-                'private_key': private_key,
-            })
-        else:
-            return GCS()
 
-    def test_service_account(self):
-        GCSTests.gcs.fget(self)
+class ParquetTestCase(unittest.TestCase):
 
-    def test_parquet2arrow(self):
-        parquet = Parquet('fixtures/gcp-data-davidkhala.dbt_davidkhala.country_codes.parquet')
-        stream_uri = "gs://davidkhala-data/gcp-data-davidkhala.dbt_davidkhala.country_codes.stream.arrow"
-        self.gcs.write_stream(stream_uri, parquet.read_stream())
-        batch_uri = "gs://davidkhala-data/gcp-data-davidkhala.dbt_davidkhala.country_codes.batch.arrow"
-        self.gcs.write_batch(batch_uri, parquet.read_batch())
+    def setUp(self):
+        self.parquet = Parquet('fixtures/gcp-data-davidkhala.dbt_davidkhala.country_codes.parquet')
 
+    def test_schema(self):
+        self.assertIsInstance(self.parquet.schema, ParquetSchema)
+        for field in self.parquet.schema:
+            self.assertIsInstance(field, ColumnSchema)
+            print(field)
+            self.assertIsInstance(field.logical_type, ParquetLogicalType)
+            self.assertEqual(str(field.logical_type), "String")
 
+    def test_stream(self):
+        for record_batch in self.parquet.read_stream():
+            print(record_batch)
+            print('<<<<')
+            for column in record_batch.column_names:
+                print(column, record_batch.column(column))
 if __name__ == '__main__':
     unittest.main()

format/tests/avro_test.py

@@ -8,11 +8,10 @@
                            )
 
 
-class Arrow2Avro:
-    arrow: Table
+class ToAvro:
 
     def __init__(self, table: Table):
-        self.arrow = table
+        self.table = table
 
     @staticmethod
     def type(arrow_type: DataType):
@@ -34,12 +33,12 @@ def type(arrow_type: DataType):
             return {"type": "int", "logicalType": "date"}
         elif is_list(arrow_type):
 
-            return {"type": "array", "items": Arrow2Avro.type(cast(ListType, arrow_type).value_type)}
+            return {"type": "array", "items": ToAvro.type(cast(ListType, arrow_type).value_type)}
         elif is_struct(arrow_type):
             return {
                 "type": "record",
                 "name": "struct",
-                "fields": [{"name": field.name, "type": Arrow2Avro.type(field.type)} for field in
+                "fields": [{"name": field.name, "type": ToAvro.type(field.type)} for field in
                            cast(StructType, arrow_type)]
             }
         else:
@@ -52,10 +51,10 @@ def schema(self):
             "name": "Root",
             "fields": list(map(lambda _field: {
                 "name": _field.name,
-                "type": Arrow2Avro.type(_field.type)
-            }, self.arrow.schema))
+                "type": ToAvro.type(_field.type)
+            }, self.table.schema))
         }
 
     @property
     def records(self):
-        return self.arrow.to_pylist()
+        return self.table.to_pylist()