Feldera Python is the Feldera SDK for Python developers.

from feldera.rest.client import Client as FelderaClient

from feldera.sql_context import SQLContext

from feldera.sql_schema import SQLSchema

from typing import Optional

from feldera.sql_schema import SQLSchema

class SQLTable:

name: str

ddl: str

schema: Optional[SQLSchema]

def __init__(self, name: str, ddl: Optional[str] = None, schema: Optional[SQLSchema] = None):

if ddl is None and schema is None:

raise ValueError("Either ddl or schema must be provided")

self.name = name

self.ddl = ddl

self.schema = schema

def build_ddl(self):

"""

if self.schema is None:

return self.ddl

if self.ddl is not None:

raise ValueError("Both ddl and schema are provided")

return self.schema.build_ddl(self.name)

import pandas as pd

from threading import Thread

from queue import Queue, Empty

from feldera import FelderaClient

from enum import Enum

class _OutputHandlerInstruction(Enum):

PipelineStarted = 1

RanToCompletion = 2

class OutputHandler(Thread):

client: FelderaClient

pipeline_name: str

view_name: str

queue: Queue

buffer: list[list[dict]] = []

def __init__(self, client: FelderaClient, pipeline_name: str, view_name: str, queue: Queue):

super().__init__()

self.client = client

self.pipeline_name = pipeline_name

self.view_name = view_name

self.queue = queue

def run(self):

"""

ack: _OutputHandlerInstruction = self.queue.get()

match ack:

case _OutputHandlerInstruction.PipelineStarted:

gen_obj = self.client.listen_to_pipeline(self.pipeline_name, self.view_name, format="json")

self.queue.task_done()

for chunk in gen_obj:

chunk: dict = chunk

data: list[dict] = chunk.get("json_data")

if data:

self.buffer.append(data)

try:

again_ack: _OutputHandlerInstruction = self.queue.get(block=False)

if again_ack:

match again_ack:

case _OutputHandlerInstruction.RanToCompletion:

self.queue.task_done()

return

case _OutputHandlerInstruction.PipelineStarted:

self.queue.task_done()

continue

except Empty:

continue

case _OutputHandlerInstruction.RanToCompletion:

self.queue.task_done()

return

def to_pandas(self):

"""

self.join()

return pd.DataFrame([

{**item['insert'], 'insert_delete': 1} if 'insert' in item else {**item['delete'], 'insert_delete': -1}

for sublist in self.buffer for item in sublist

])

from feldera.rest.client import Client

import logging

from feldera.rest.config import Config

from feldera.rest.errors import FelderaAPIError, FelderaTimeoutError, FelderaCommunicationError

import json

import requests

from typing import Callable, Optional, Any, Union, Mapping, Sequence, List

class HttpRequests:

def __init__(self, config: Config) -> None:

self.config = config

self.headers = {

"User-Agent": "feldera-python-sdk/v1"

}

if self.config.api_key:

self.headers["Authorization"] = f"Bearer {self.config.api_key}"

def send_request(

self,

http_method: Callable,

path: str,

body: Optional[

Union[Mapping[str, Any], Sequence[Mapping[str, Any]], List[str], str]

] = None,

content_type: str = "application/json",

params: Optional[Mapping[str, Any]] = None,

stream: bool = False,

) -> Any:

"""

self.headers["Content-Type"] = content_type

try:

timeout = self.config.timeout

headers = self.headers

request_path = self.config.url + "/" + self.config.version + path

logging.debug(

"sending %s request to: %s with headers: %s, and params: %s",

http_method.__name__, request_path, str(headers), str(params)

)

if http_method.__name__ == "get":

request = http_method(

request_path,

timeout=timeout,

headers=headers,

params=params,

)

elif isinstance(body, bytes):

request = http_method(

request_path,

timeout=timeout,

headers=headers,

data=body,

params=params,

stream=stream,

)

else:

request = http_method(

request_path,

timeout=timeout,

headers=headers,

data=json.dumps(body) if body else "" if body == "" else "null",

params=params,

stream=stream,

)

if stream:

return request

resp = self.__validate(request)

logging.debug("got response: %s", str(resp))

return resp

except requests.exceptions.Timeout as err:

raise FelderaTimeoutError(str(err)) from err

except requests.exceptions.ConnectionError as err:

raise FelderaCommunicationError(str(err)) from err

def get(

self,

path: str,

params: Optional[Mapping[str, Any]] = None

) -> Any:

return self.send_request(requests.get, path, params)

def post(

self,

path: str,

body: Optional[

Union[Mapping[str, Any], Sequence[Mapping[str, Any]], List[str], str]

] = None,

content_type: Optional[str] = "application/json",

params: Optional[Mapping[str, Any]] = None,

stream: bool = False,

) -> Any:

return self.send_request(requests.post, path, body, content_type, params, stream=stream)

def patch(

self,

path: str,

body: Optional[

Union[Mapping[str, Any], Sequence[Mapping[str, Any]], List[str], str]

] = None,

content_type: Optional[str] = "application/json",

params: Optional[Mapping[str, Any]] = None

) -> Any:

return self.send_request(requests.patch, path, body, content_type, params)

def put(

self,

path: str,

body: Optional[

Union[Mapping[str, Any], Sequence[Mapping[str, Any]], List[str], str]

] = None,

content_type: Optional[str] = "application/json",

params: Optional[Mapping[str, Any]] = None

) -> Any:

return self.send_request(requests.put, path, body, content_type, params)

def delete(

self,

path: str,

body: Optional[Union[Mapping[str, Any], Sequence[Mapping[str, Any]], List[str]]] = None,

params: Optional[Mapping[str, Any]] = None

) -> Any:

return self.send_request(requests.delete, path, body, params=params)

@staticmethod

def __to_json(request: requests.Response) -> Any:

if request.content == b"":

return request

return request.json()

@staticmethod

def __validate(request: requests.Response) -> Any:

try:

request.raise_for_status()

resp = HttpRequests.__to_json(request)

return resp

except requests.exceptions.HTTPError as err:

raise FelderaAPIError(str(err), request) from err

import uuid

from typing import Optional

class AttachedConnector:

"""

name: str

is_input: bool

connector_name: str

relation_name: str

def __init__(self, connector_name: str, relation_name: str, is_input: bool, name: Optional[str] = None):

self.name = name or str(uuid.uuid4())

self.is_input = is_input

self.connector_name = connector_name

self.relation_name = relation_name

def to_json(self):

return {

"name": self.name,

"is_input": self.is_input,

"connector_name": self.connector_name,

"relation_name": self.relation_name

}