Cursor Rules Template: Centrifugo Realtime Messaging with Python

Overview

Cursor Rules Template for Centrifugo realtime messaging in Python. It defines a precise role for Cursor AI to guide development of a Python backend using FastAPI and Centrifugo. It includes a copyable .cursorrules configuration and a recommended project structure for production-ready apps.

• Direct answer: Paste the .cursorrules block into your repository root; it configures the AI to produce safe, testable, production-ready code for Centrifugo based realtime messaging with Python.
• Stack coverage: Python, FastAPI, Centrifugo, SQLAlchemy, Pytest, and CI/CD.

When to Use These Cursor Rules

• When building a Python backend with Centrifugo for real-time messaging.
• When you want a repeatable, copyable Cursor configuration to enforce structure.
• When you need testing and linting integrated into the workflow.
• When you want to stabilize authentication and channel security in production.

Copyable .cursorrules Configuration

.cursorrules
FRAMEWORK: Python FastAPI
CTX: Centrifugo real-time messaging; async backend; WebSocket clients
ROLE: Backend service coordinating messages between Python FastAPI endpoints and Centrifugo channels
CODE_STYLE: PEP8, Black, isort, mypy
ARCHITECTURE: [
  project_root/
  app/
  app/api/
  app/models/
  app/services/
  centrifugo/
  tests/
  requirements.txt
  Dockerfile
]
AUTH: OAuth2 with JWTs; validate token; short-lived access tokens; rotate tokens; verify origin
DB: SQLAlchemy async; PostgreSQL; models in app/models; migrations via Alembic
TESTS_LINT: pytest, pytest-asyncio; black --check; ruff; mypy --strict; pre-commit
PROHIBITED_ACTIONS: Do not bypass auth; Do not perform direct DB writes from AI; Do not generate secrets in code; Do not assume insecure defaults

Recommended Project Structure

project_root/
├── app/
│   ├── main.py
│   ├── api/
│   │   └── endpoints.py
│   ├── models/
│   │   └── message.py
│   ├── services/
│   │   └── centrifugo_client.py
│   ├── core/
│   │   └── config.py
│   └── schemas/
│       └── message.py
├── centrifugo/
│   └── centrifugo.json
├── tests/
│   ├── test_messaging.py
│   └── conftest.py
├── requirements.txt
└── Dockerfile

Core Engineering Principles

• Explicit configuration over magic behavior to prevent drift between AI guidance and runtime behavior.
• Separation of concerns across API, messaging, and persistence layers.
• Validate all inputs and constrain real-time payloads to known schemas.
• Secure by default: least privilege access, token-based auth, encrypted channels.
• Observability: structured logging, traces, metrics for message flows.
• Testability: unit tests for business logic and integration tests for messaging.
• Performance: prefer asynchronous I/O and connection pooling for Centrifugo interactions.

Code Construction Rules

• Use a strict Python typing discipline with pydantic models for payload validation.
• Place Centrifugo client logic in app/services/centrifugo_client.py with clear interface.
• Publish and subscribe to channels asynchronously; do not block the event loop.
• Follow a consistent naming convention for endpoints, services, and models.
• Do not embed credentials or secrets in code. Use environment variables and secrets management.
• Do not bypass authentication checks when publishing to or subscribing from Centrifugo.
• Keep the .cursorrules grounded in real stack constraints and avoid speculative features.

Security and Production Rules

• Enforce TLS in all external communications and verify Centrifugo server certificates.
• Use OAuth2 with short-lived JWTs for API clients and WebSocket connections; rotate keys regularly.
• Validate origin and channel permissions before publishing to Centrifugo channels.
• Mask and avoid logging sensitive payloads; redact tokens and secrets.
• Isolate Centrifugo in its own network segment; restrict topics and access controls.

Testing Checklist

• Unit tests for the Python domain logic and message serializers.
• Integration tests that mock Centrifugo or run a local test instance.
• End-to-end tests with WebSocket clients to exercise publish/subscribe flows.
• Linting and type checks in CI: black, ruff, and mypy.
• Security tests for token validation, channel permissions, and input validation.

Common Mistakes to Avoid

• Assuming Centrifugo channels are publicly accessible without ACLs.
• Weak token handling or long-lived credentials stored in code.
• Blocking the event loop with synchronous DB calls or I/O.
• Overly permissive topic subscriptions or lacking input validation.
• Skipping migrations or not keeping test environments in sync with production.

FAQ

How do I initialize Centrifugo with Python FastAPI?

Initialize a Centrifugo client in app/services/centrifugo_client.py; configure the client with the Centrifugo server URL, API key, and an async event loop. Use FastAPI endpoints to publish messages and WebSocket clients to subscribe. Cursor AI rules guide the integration and security constraints.

How do I publish messages to Centrifugo channels from FastAPI endpoints?

Expose an endpoint that validates input payloads against a pydantic model, then call the Centrifugo client to publish to a specific channel. Ensure the client handles concurrency and errors gracefully, and secures the channel with appropriate permissions.

How should authentication be configured for Centrifugo channels in Python?

Use OAuth2 with JWT for API clients, and generate per-session tokens for WebSocket clients. Validate the token on publish and subscribe requests, and enforce channel-level permissions in the backend and in Centrifugo config.

How can I test real-time messaging with pytest?

Write unit tests for serialization and channel logic, integration tests with a mock Centrifugo server, and end-to-end tests using WebSocket clients. Use pytest-asyncio to drive async code paths and simulate publish/subscribe flows.

How do I deploy Centrifugo and the Python service in production?

Deploy Centrifugo in a dedicated container, secure its endpoints with TLS, and run the Python service behind a reverse proxy with TLS termination. Use environment-based configuration, a robust CI/CD pipeline, and automated tests to validate messaging correctness before release.