feat(reporting): custom extractor for bitfinex API

meltano/extract/tap-bitfinexapi/.gitignore

@@ -0,0 +1 @@
+*/__pycache__/

meltano/extract/tap-bitfinexapi/README.md

@@ -0,0 +1,75 @@
+# tap-bitfinexapi
+
+`tap-bitfinexapi` is a Singer tap for BitfinexApi.
+
+Built with the [Meltano Tap SDK](https://sdk.meltano.com) for Singer Taps.
+
+## Configuration
+
+### Accepted Config Options
+
+<!--
+Developer TODO: Provide a list of config options accepted by the tap.
+
+This section can be created by copy-pasting the CLI output from:
+
+```
+tap-bitfinexapi --about --format=markdown
+```
+-->
+
+A full list of supported settings and capabilities for this
+tap is available by running:
+
+```bash
+tap-bitfinexapi --about
+```
+
+### Configure using environment variables
+
+This Singer tap will automatically import any environment variables within the working directory's
+`.env` if the `--config=ENV` is provided, such that config values will be considered if a matching
+environment variable is set either in the terminal context or in the `.env` file.
+
+### Source Authentication and Authorization
+
+<!--
+Developer TODO: If your tap requires special access on the source system, or any special authentication requirements, provide those here.
+-->
+
+## Usage
+
+You can easily run `tap-bitfinexapi` by itself or in a pipeline using [Meltano](https://meltano.com/).
+
+### Executing the Tap Directly
+
+```bash
+tap-bitfinexapi --version
+tap-bitfinexapi --help
+tap-bitfinexapi --config CONFIG --discover > ./catalog.json
+```
+
+### Testing with [Meltano](https://www.meltano.com)
+
+_**Note:** This tap will work in any Singer environment and does not require Meltano.
+Examples here are for convenience and to streamline end-to-end orchestration scenarios._
+
+Next, install Meltano (if you haven't already) and any needed plugins:
+
+```bash
+# Install meltano
+pipx install meltano
+# Initialize meltano within this directory
+cd tap-bitfinexapi
+meltano install
+```
+
+Now you can test and orchestrate using Meltano:
+
+```bash
+# Test invocation:
+meltano invoke tap-bitfinexapi --version
+
+# OR run a test ELT pipeline:
+meltano run tap-bitfinexapi target-jsonl
+```

meltano/extract/tap-bitfinexapi/pyproject.toml

@@ -0,0 +1,70 @@
+[project]
+name = "tap-bitfinexapi"
+version = "0.0.1"
+description = "Singer tap for BitfinexApi, built with the Meltano Singer SDK."
+readme = "README.md"
+authors = [{ name = "Jose Rojas Echenique", email = "[email protected]" }]
+keywords = [
+    "ELT",
+    "BitfinexApi",
+]
+classifiers = [
+    "Intended Audience :: Developers",
+    "License :: OSI Approved :: Apache Software License",
+    "Operating System :: OS Independent",
+    "Programming Language :: Python :: 3.9",
+    "Programming Language :: Python :: 3.10",
+    "Programming Language :: Python :: 3.11",
+    "Programming Language :: Python :: 3.12",
+    "Programming Language :: Python :: 3.13",
+]
+license = "Apache-2.0"
+license-files = [ "LICENSE" ]
+requires-python = ">=3.9"
+dynamic = ["dependencies"]
+
+[tool.poetry]
+
+[tool.poetry.dependencies]
+singer-sdk = { version="~=0.43.1", extras = [] }
+fs-s3fs = { version = "~=1.1.1", optional = true }
+requests = "~=2.32.3"
+
+[tool.poetry.group.dev.dependencies]
+pytest = ">=8"
+singer-sdk = { version="~=0.43.1", extras = ["testing"] }
+
+[tool.poetry.extras]
+s3 = ["fs-s3fs"]
+
+[tool.pytest.ini_options]
+addopts = [
+    "--durations=10",
+]
+
+[tool.mypy]
+python_version = "3.12"
+warn_unused_configs = true
+
+[tool.ruff]
+target-version = "py39"
+
+[tool.ruff.lint]
+ignore = [
+    "COM812",  # missing-trailing-comma
+]
+select = ["ALL"]
+
+[tool.ruff.lint.flake8-annotations]
+allow-star-arg-any = true
+
+[tool.ruff.lint.pydocstyle]
+convention = "google"
+
+[build-system]
+requires = ["poetry-core>=2,<3"]
+build-backend = "poetry.core.masonry.api"
+
+[tool.poetry.scripts]
+# CLI declaration
+tap-bitfinexapi = 'tap_bitfinexapi.tap:TapBitfinexApi.cli'

meltano/extract/tap-bitfinexapi/tap_bitfinexapi/__init__.py

@@ -0,0 +1,4 @@
+"""Tap for BitfinexApi."""
+from .streams import TickerStream
+
+__all__ = [TickerStream]

meltano/extract/tap-bitfinexapi/tap_bitfinexapi/__main__.py

@@ -0,0 +1,7 @@
+"""BitfinexApi entry point."""
+
+from __future__ import annotations
+
+from tap_bitfinexapi.tap import TapBitfinexApi
+
+TapBitfinexApi.cli()

meltano/extract/tap-bitfinexapi/tap_bitfinexapi/client.py

@@ -0,0 +1,40 @@
+"""REST client handling, including BitfinexApiStream base class."""
+
+from __future__ import annotations
+
+import decimal
+import typing as t
+from importlib import resources
+
+from singer_sdk.helpers.jsonpath import extract_jsonpath
+from singer_sdk.pagination import BaseAPIPaginator  # noqa: TC002
+from singer_sdk.streams import RESTStream
+
+if t.TYPE_CHECKING:
+    import requests
+    from singer_sdk.helpers.types import Context
+
+
+class BitfinexApiStream(RESTStream):
+    """BitfinexApi stream class."""
+
+    @property
+    def url_base(self) -> str:
+        """Return the API URL root, configurable via tap settings."""
+        # TODO: hardcode a value here, or retrieve it from self.config
+        return "https://api-pub.bitfinex.com"
+
+    def parse_response(self, response: requests.Response) -> t.Iterable[dict]:
+        """Parse the response and return an iterator of result records.
+
+        Args:
+            response: The HTTP ``requests.Response`` object.
+
+        Yields:
+            Each record from the source.
+        """
+
+        yield dict(zip(
+		["BID", "BID_SIZE", "ASK", "ASK_SIZE", "DAILY_CHANGE", "DAILY_CHANGE_RELATIVE", "LAST_PRICE", "VOLUME", "HIGH", "LOW"],
+		response.json(parse_float=decimal.Decimal),
+	))

meltano/extract/tap-bitfinexapi/tap_bitfinexapi/streams.py

@@ -0,0 +1,71 @@
+"""Stream type classes for tap-bitfinexapi."""
+
+from __future__ import annotations
+
+import typing as t
+from importlib import resources
+
+from singer_sdk import typing as th  # JSON Schema typing helpers
+
+from tap_bitfinexapi.client import BitfinexApiStream
+
+class TickerStream(BitfinexApiStream):
+    """Ticker
+    https://docs.bitfinex.com/reference/rest-public-ticker
+    """
+    name = "ticker"
+    path = "/v2/ticker/tBTCUSD"
+    primary_keys: t.ClassVar[list[str]] = ["requested_at"]
+    schema = th.PropertiesList(
+        th.Property("requested_at", th.DateTimeType),
+        th.Property(
+            "BID",
+            th.NumberType,
+            description="Price of last highest bid",
+        ),
+        th.Property(
+            "BID_SIZE",
+            th.NumberType,
+            description="Sum of the 25 highest bid sizes",
+        ),
+        th.Property(
+            "ASK",
+            th.NumberType,
+            description="Price of last lowest ask",
+        ),
+        th.Property(
+            "ASK_SIZE",
+            th.NumberType,
+            description="Sum of the 25 lowest ask sizes",
+        ),
+        th.Property(
+            "DAILY_CHANGE",
+            th.NumberType,
+            description="Amount that the last price has changed since yesterday",
+        ),
+        th.Property(
+            "DAILY_CHANGE_RELATIVE",
+            th.NumberType,
+            description="Relative price change since yesterday (*100 for percentage change)",
+        ),
+        th.Property(
+            "LAST_PRICE",
+            th.NumberType,
+            description="Price of the last trade",
+        ),
+        th.Property(
+            "VOLUME",
+            th.NumberType,
+            description="Daily volume",
+        ),
+        th.Property(
+            "HIGH",
+            th.NumberType,
+            description="Daily high",
+        ),
+        th.Property(
+            "LOW",
+            th.NumberType,
+            description="Daily low",
+        ),
+    ).to_dict()

meltano/extract/tap-bitfinexapi/tap_bitfinexapi/tap.py

@@ -0,0 +1,43 @@
+"""BitfinexApi tap class."""
+
+from __future__ import annotations
+
+from singer_sdk import Tap
+from singer_sdk import typing as th  # JSON schema typing helpers
+
+from tap_bitfinexapi import TickerStream
+
+STREAM_TYPES = [TickerStream]
+
+class TapBitfinexApi(Tap):
+    """BitfinexApi tap class."""
+
+    name = "tap-bitfinexapi"
+
+    # TODO: Update this section with the actual config values you expect:
+    config_jsonschema = th.PropertiesList(
+        th.Property(
+            "api_url",
+            th.StringType,
+            title="API URL",
+            default="https://api.mysample.com",
+            description="The url for the API service",
+        ),
+        th.Property(
+            "user_agent",
+            th.StringType,
+            description=(
+                "A custom User-Agent header to send with each request. Default is "
+                "'<tap_name>/<tap_version>'"
+            ),
+        ),
+    ).to_dict()
+
+    def discover_streams(self) -> list[streams.BitfinexApiStream]:
+        """Return a list of discovered streams.
+        """
+        return [stream_class(tap=self) for stream_class in STREAM_TYPES]
+
+
+if __name__ == "__main__":
+    TapBitfinexApi.cli()

meltano/meltano.yml

@@ -25,6 +25,10 @@ plugins:
       flattening_max_depth: 10
     select:
     - public-*.*
+  - name: tap-bitfinexapi
+    namespace: tap_bitfinexapi
+    pip_url: -e extract/tap-bitfinexapi
+    executable: tap-bitfinexapi
   loaders:
   - name: target-bigquery
     variant: z3z1ma
@@ -54,3 +58,11 @@ plugins:
     - name: project
       env: DBT_BIGQUERY_PROJECT
       value: lana-dev-440721
+jobs:
+- name: poll-bitfinex
+  tasks:
+  - tap-bitfinexapi target-bigquery
+schedules:
+- name: poll-bitfinex-on-minute
+  interval: '* * * * *'
+  job: poll-bitfinex