Verdin is a tiny bird, and also a Tinybird SDK for Python.
pip install verdin Python 3.10+
# the tinybird module exposes all important tinybird conceptsfromverdinimporttinybirdclient=tinybird.Client("p.mytoken") query=client.sql("select * from my_datasource__v0") # run the query with `FORMAT JSON` and receive a QueryJsonResultresponse: tinybird.QueryJsonResult=query.json() # print records returned from the pipeprint(response.data)You can also run, e.g., query.get(format=OutputFormat.CSV) to get the raw response with CSV data.
fromverdinimporttinybirdclient=tinybird.Client("p.mytoken") pipe=client.pipe("my_pipe") # query the pipe using dynamic parametersresponse: tinybird.PipeJsonResponse=pipe.query({"key": "val"}) # print records returned from the pipeprint(response.data)fromverdinimporttinybirdclient=tinybird.Client("p.mytoken") # will access my_datasource__v0datasource=client.datasource("my_datasource", version=0) # query the pipe using dynamic parametersdatasource.append([ ("col1-row1", "col2-row1"), ("col1-row2", "col2-row2"), ])The DataSource object also gives you access to /v0/events, which is the high-frequency ingest, to append data. Use the send_events method and pass JSON serializable documents to it.
datasource.send_events(records=[{"key": "val1"},{"key": "val2"}, ... ])Verdin provides a way to queue and batch data continuously:
fromqueueimportQueuefromthreadingimportThreadfromverdinimporttinybirdfromverdin.workerimportQueuingDatasourceAppenderclient=tinybird.Client("p.mytoken") records=Queue() appender=QueuingDatasourceAppender(records, client.datasource("my_datasource")) Thread(target=appender.run).start() # appender will regularly read batches of data from the queue and append them# to the datasource. the appender respects rate limiting.records.put(("col1-row1", "col2-row1")) records.put(("col1-row2", "col2-row2"))The DataSource and Pipes objects presented so far are high-level abstractions that provide a convenience Python API to deal with the most common use cases. Verdin also provides more low-level access to APIs via client.api. The following APIs are available:
/v0/datasources:client.api.datasources/v0/events:client.api.events/v0/pipes:client.api.pipes/v0/sql:client.api.query/v0/tokens:client.api.tokens/v0/variables:client.api.variables
Note that for some (datasources, pipes, tokens), manipulation operations are not implemented as they are typically done through tb deployments and not through the API.
Also note that API clients do not take care of retries or rate limiting. The caller is expected to handle fault tolerance.
You can query a pipe through the pipes API as follows:
fromverdinimporttinybirdclient=tinybird.Client(...) response=client.api.pipes.query( "my_pipe", parameters={"my_param": "..."}, query="SELECT * FROM _ LIMIT 10", ) forrecordinresponse.data: # each record is a dictionary ...You can use the HFI endpoint /v0/events through the events api. As records, you can pass a list of JSON serializable documents.
fromverdinimporttinybirdclient=tinybird.Client(...) response=client.api.events.send("my_datasource", records=[{"id": "...", "value": "..."}, ... ]) assertresponse.quarantined_rows==0Create the virtual environment, install dependencies, and run tests
make venv make test Run the code formatter
make format Upload the pypi package using twine
make upload