A professional FastAPI + Celery Robotic Process Automation (RPA) bootstrap project. Designed for reliability, observability, and horizontal scalability.
Tip
This repository is a template. Use it to jumpstart your own RPA developments by implementing the RpaCoreAbstract class.
Most if not all parts were vibe coded, like what modern devs do 😏
| Screenshot 1 | Screenshot 2 | Screenshot 3 |
|---|---|---|
 |
 |
 |
The system follows a decoupled architecture where the API handles submissions, and independent workers process tasks using Selenium/Chrome.
graph TD
Caller[External Caller] -- POST /submit-task --> API[FastAPI]
Caller -- batch uploads --> SQL
API -- writes --> SQL[(DB)]
Beat[Celery Beat] -- claims rows --> SQL
Beat -- dispatch --> Redis[Redis Broker]
Redis -- consume --> Worker[Celery Worker]
Worker -- Selenium --> Web[Target Website]
Worker -- HTTP POST --> Callback[Callback Hook]
- Atomic Task Claiming: Uses
FOR UPDATE SKIP LOCKED(Postgres) or a safe SELECT-UPDATE cycle (SQLite) to ensure no row is ever double-processed. - Fail-Safe Execution: Tasks stay in Redis until acknowledged (
task_acks_late=True). Crashed workers result in automatically re-queued tasks. - Observability: Structured JSON logging with end-to-end correlation IDs across all services.
- Flow Control:
worker_prefetch_multiplier=1ensures workers only take what they can handle right now.
- Python 3.11+
- PostgreSQL & Redis (Local or Remote)
- Google Chrome (ChromeDriver is managed automatically)
# Clone and enter directory
git clone https://github.com/DonnC/rpa-core.git
cd rpa-core
# Create virtual environment
python -m venv .venv
source .venv/bin/activate # Windows: .venv\Scripts\activate
# Install dependencies
pip install -r requirements.txt
# Configure environment
cp .env.example .env
# Edit .env with your DB and Redis credentialsalembic upgrade headTo create your own automation, inherit from RpaCoreAbstract in app/rpa_impl/.
- Define Logic: Implement the
process()method. - Use WebDriver: Use
self.get_driver()for a pre-configured Selenium instance. - Checkpointing: For long tasks, call
self.save_checkpoint()after each major step.
from app.rpa_core import RpaCoreAbstract
class MyCustomBot(RpaCoreAbstract):
def process(self, task, db, **kwargs):
driver = self.get_driver()
driver.get(task.payload["url"])
# ... your automation logic ...
return {"result": "Success!", "data": {"extracted_text": "..."}}You need four active processes. For convenience, use the management scripts:
| Platform | Command |
|---|---|
| Windows | .\manage.ps1 |
| Linux/macOS | ./manage.sh |
If you prefer separate terminals:
- API:
uvicorn app.main:app --reload - Worker:
celery -A app.celery_app worker --loglevel=info - Beat:
celery -A app.celery_app beat --loglevel=info - Flower:
celery -A app.celery_app flower --port=5555
Access at http://localhost:5555. Ideal for real-time task status, worker health, and retry monitoring.
Logs are written to logs/ in JSON format.
# Tail worker logs with colorized JSON (requires jq)
tail -f logs/worker.log | jq .The project includes a comprehensive test suite covering the API, DB claiming logic, and task flow orchestration.
# Run all tests
pytest app/tests/ -v
# Run with coverage report
pytest --cov=app app/tests/Before deploying to production, ensure these items are checked:
- TLS: Run FastAPI behind a reverse proxy (Nginx/Caddy) with HTTPS.
- Auth: Enable API Key or OAuth2 protection on endpoints.
- Scaling: Tune
WORKER_CONCURRENCYbased on your server RAM/CPU. - Persistence: Enable Redis AOF to prevent queue loss on restart.
- Monitoring: Integrate Sentry (configurable in
logging_config.py).
Distributed under the MIT License. See LICENSE for more information.