Revert "feat: bigquery_fetcher entrypoint (#67)"

.pre-commit-config.yaml

@@ -34,7 +34,6 @@ repos:
         additional_dependencies: [
           pytest==7.1.2,
           sentry-arroyo==2.22.0,
-          google-cloud-bigquery==3.41.0,
         ]
         files: ^py/.+
 default_language_version:

Dockerfile

@@ -2,8 +2,7 @@ FROM us-docker.pkg.dev/sentryio/dhi-mirror/python:3.13-debian13-dev AS build
 
 COPY py/ /app/py/
 
-# Install `bigquery` extra to enable the `bigquery_fetcher` entrypoint
-RUN python3 -m venv /opt/venv && /opt/venv/bin/pip install --no-cache-dir "/app/py[bigquery]"
+RUN python3 -m venv /opt/venv && /opt/venv/bin/pip install --no-cache-dir /app/py
 
 FROM us-docker.pkg.dev/sentryio/dhi-mirror/python:3.13-debian13
 

py/pyproject.toml

@@ -16,10 +16,6 @@ dependencies = [
     'sentry-arroyo>=2.22.0'
 ]
 
-[project.optional-dependencies]
-# Only needed for the `bigquery_fetcher` entrypoint
-bigquery = ['google-cloud-bigquery>=3.0.0']
-
 [tool.setuptools.package-data]
 usageaccountant = ["py.typed"]
 

py/requirements-test.txt

@@ -1,6 +1,5 @@
 sentry-arroyo>=2.22.0
 pytest>=7.2.1
 mypy>=1.4.1
-google-cloud-bigquery>=3.0.0
 black>=24.10.0
 flake8>=6

py/tests/test_fetcher.py

@@ -1,4 +1,5 @@
 import copy
+import os
 import unittest
 from io import StringIO
 from unittest.mock import Mock, patch
@@ -13,7 +14,6 @@
 from tests.test_data import datadog_response
 from usageaccountant import accumulator
 from usageaccountant import datadog_fetcher as ddf
-from usageaccountant.fetcher_utils import UsageAccumulatorRecord
 
 
 class TestDatadogFetcher(unittest.TestCase):
@@ -22,6 +22,12 @@ class TestDatadogFetcher(unittest.TestCase):
         'by {app_feature,shared_resource_id}","unit": "bytes",'
         '"shared_resource_id": "rc_long_redis"}]'
     )
+    kafka_config_str = (
+        '{"bootstrap_servers": ["kafka.service.host:1234"],'
+        '"config_params": '
+        '{"compression.type": "lz4",'
+        '"message.max.bytes": 10}}'
+    )
 
     def setUp(self) -> None:
         self.good_response = copy.deepcopy(datadog_response.good_response)
@@ -56,6 +62,40 @@ def test_parse_and_assert_query_file_no_query(self) -> None:
             AssertionError, ddf.parse_and_assert_query_file, query_io
         )
 
+    def test_parse_and_assert_kafka_config(self) -> None:
+        kafka_io = StringIO(self.kafka_config_str)
+        ddf.parse_and_assert_kafka_config(kafka_io)
+
+    def test_parse_and_assert_kafka_config_no_config_params(self) -> None:
+        kafka_io = StringIO(
+            '{"bootstrap_servers": ["kafka.service.host:1234"]}'
+        )
+        self.assertRaises(
+            AssertionError, ddf.parse_and_assert_kafka_config, kafka_io
+        )
+
+    def test_parse_and_assert_expandvars(self) -> None:
+        os.environ["SASL_PASSWORD"] = "supersecret"
+
+        expected_kafka_config = accumulator.KafkaConfig(
+            bootstrap_servers=["kafka.service.host:1234"],
+            config_params={
+                "sasl.username": "alice",
+                "sasl.password": "supersecret",
+            },
+        )
+
+        kafka_io = StringIO(
+            '{"bootstrap_servers": ["kafka.service.host:1234"],'
+            '"config_params": '
+            '{"sasl.username": "alice",'
+            '"sasl.password": "${SASL_PASSWORD}"}}'
+        )
+
+        self.assertEqual(
+            ddf.parse_and_assert_kafka_config(kafka_io), expected_kafka_config
+        )
+
     def test_parse_and_assert_unit(self) -> None:
         parsed_unit = ddf.parse_and_assert_unit(
             cast(
@@ -148,19 +188,17 @@ def test_parse_and_assert_response_scope_no_app_feature(self) -> None:
 
     def test_process_series_data(self) -> None:
         expected_record_list = [
-            UsageAccumulatorRecord(
+            ddf.UsageAccumulatorRecord(
                 resource_id="rc_long_redis",
                 app_feature="shared",
                 amount=2,
                 usage_type=accumulator.UsageUnit.BYTES,
-                timestamp=None,
             ),
-            UsageAccumulatorRecord(
+            ddf.UsageAccumulatorRecord(
                 resource_id="rc_long_redis",
                 app_feature="shared",
                 amount=3,
                 usage_type=accumulator.UsageUnit.BYTES,
-                timestamp=None,
             ),
         ]
 
@@ -178,12 +216,11 @@ def test_process_series_data(self) -> None:
 
     def test_process_series_data_with_none(self) -> None:
         expected_record_list = [
-            UsageAccumulatorRecord(
+            ddf.UsageAccumulatorRecord(
                 resource_id="rc_long_redis",
                 app_feature="shared",
                 amount=2,
                 usage_type=accumulator.UsageUnit.BYTES,
-                timestamp=None,
             )
         ]
 
@@ -201,19 +238,17 @@ def test_process_series_data_with_none(self) -> None:
 
     def test_process_series_data_feature_tag(self) -> None:
         expected_record_list = [
-            UsageAccumulatorRecord(
+            ddf.UsageAccumulatorRecord(
                 resource_id="rc_long_redis",
                 app_feature="shared",
                 amount=2,
                 usage_type=accumulator.UsageUnit.BYTES,
-                timestamp=None,
             ),
-            UsageAccumulatorRecord(
+            ddf.UsageAccumulatorRecord(
                 resource_id="rc_long_redis",
                 app_feature="shared",
                 amount=3,
                 usage_type=accumulator.UsageUnit.BYTES,
-                timestamp=None,
             ),
         ]
         parsed = ddf.parse_and_assert_response_series(