Python library and CLI tool for polling events from the Chaturbate API featuring asynchronous event handling, structured logging, and optional InfluxDB integration for analytics and monitoring.
- Event Polling: Retrieve real-time events from the Chaturbate API.
- Error Handling: Built-in retries, exponential backoff, and error classification.
- Logging: Supports structured JSON logs and console outputs for better debugging.
- Optional InfluxDB Integration: Store and analyze events using InfluxDB.
Ensure Python 3.11 or later is installed, then install the package via pip:
pip install chaturbate-poller
Create a .env
file in your project's root directory with the following:
CB_USERNAME="your_chaturbate_username"
CB_TOKEN="your_chaturbate_token"
INFLUXDB_URL="http://influxdb:8086"
INFLUXDB_TOKEN="your_influxdb_token"
INFLUXDB_ORG="chaturbate-poller"
INFLUXDB_BUCKET="my-bucket"
USE_DATABASE="false" # Set to `true` if InfluxDB is used
💡 Tip: Generate an API token here with "Events API" permission enabled.
Start the poller with the following command:
python -m chaturbate_poller start --username <your_username> --token <your_token>
--username
: Your Chaturbate username. Defaults to.env
file value.--token
: Your API token. Defaults to.env
file value.--timeout
: Timeout for API requests (default: 10 seconds).--database
: Enable InfluxDB integration. Defaults to disabled.--testbed
: Enable the testbed environment for testing.--verbose
: Enable detailed logging for debugging.
Run python -m chaturbate_poller --help
for a complete list of options.
To run the poller in Docker, pull the image and start the container:
docker pull ghcr.io/mountaingod2/chaturbate_poller:latest
docker run \
-e CB_USERNAME="your_chaturbate_username" \
-e CB_TOKEN="your_chaturbate_token" \
ghcr.io/mountaingod2/chaturbate_poller:latest --verbose --database
The library can also be used directly in your Python code:
import asyncio
from chaturbate_poller import ChaturbateClient
async def main():
async with ChaturbateClient("your_username", "your_token", testbed=False) as client:
url = None
while True:
response = await client.fetch_events(url)
for event in response.events:
# Do something with the event
print(event.model_dump())
url = response.next_url
if __name__ == "__main__":
asyncio.run(main())
-
Clone the repository:
git clone https://github.com/MountainGod2/chaturbate_poller.git cd chaturbate_poller
-
Install dependencies using uv:
uv sync --all-extras
Run tests with pytest
:
uv run pytest
Build and preview the documentation locally:
uv sync --extra=docs
uv run make clean html -C ./docs
Contributions are welcome! Here's how to get started:
- Fork the repository.
- Create a feature branch.
- Submit a pull request, ensuring it includes tests and adheres to the coding standards.
This project is licensed under the MIT License. See the LICENSE file for details.