Cursor rule

1---
2globs:
3  - "packages/hermes-api/app/db/**/*.py"
4---
5 
6# Hermes API - Database Layer Rules
7 
8## SQLAlchemy Async Patterns
9 
10### Session Management
11- Always use async sessions: `AsyncSession`
12- Get sessions via dependency injection: `get_database_session()`
13- Never create sessions directly - use the session factory
14- Sessions are automatically committed/rolled back by FastAPI
15 
16Example:
17```python
18from sqlalchemy.ext.asyncio import AsyncSession
19from app.db.session import get_database_session
20 
21async def my_endpoint(
22    db: AsyncSession = Depends(get_database_session)
23):
24    # Use db session
25    pass
26```
27 
28### Queries
29- Use async query methods: `execute()`, `scalar()`, `scalars()`
30- Always `await` database operations
31- Use `select()` for queries, not legacy query API
32 
33Example:
34```python
35from sqlalchemy import select
36from app.db.models import User
37 
38result = await session.execute(
39    select(User).where(User.username == username)
40)
41user = result.scalar_one_or_none()
42```
43 
44## Repository Pattern
45 
46### Structure
47- Define repositories in `app/db/repositories.py`
48- Each model should have a corresponding repository class
49- Repositories encapsulate data access logic
50- Keep business logic out of repositories
51 
52### Repository Methods
53- Use descriptive method names: `get_by_id()`, `create()`, `update()`, `delete()`
54- Return model instances or None (not query objects)
55- Handle exceptions at repository level
56- Use type hints for all parameters and returns
57 
58Example:
59```python
60class UserRepository:
61    def __init__(self, session: AsyncSession):
62        self.session = session
63    
64    async def get_by_id(self, user_id: str) -> Optional[User]:
65        result = await self.session.execute(
66            select(User).where(User.id == user_id)
67        )
68        return result.scalar_one_or_none()
69    
70    async def create(self, user_data: dict) -> User:
71        user = User(**user_data)
72        self.session.add(user)
73        await self.session.flush()
74        return user
75```
76 
77### Repository Access
78- Access repositories via `get_repositories()` function
79- Returns dictionary of repository instances
80- Repositories share the same session
81 
82```python
83from app.db.repositories import get_repositories
84 
85repos = await get_repositories()
86user = await repos["users"].get_by_id(user_id)
87```
88 
89## Models
90 
91### Model Definition
92- Define models in `app/db/models.py`
93- Inherit from declarative base defined in `app/db/base.py`
94- Use type hints in column definitions
95- Include `__tablename__` attribute
96 
97### Column Definitions
98- Use appropriate SQLAlchemy types
99- Set `nullable=False` for required fields
100- Define `default` or `server_default` for defaults
101- Use `index=True` for frequently queried columns
102 
103### Relationships
104- Define relationships using `relationship()`
105- Use `back_populates` for bidirectional relationships
106- Use `lazy="selectin"` for async loading
107- Consider cascade delete behavior
108 
109Example:
110```python
111from sqlalchemy.orm import Mapped, mapped_column, relationship
112from datetime import datetime
113from typing import Optional
114 
115class User(Base):
116    __tablename__ = "users"
117    
118    id: Mapped[str] = mapped_column(primary_key=True)
119    username: Mapped[str] = mapped_column(unique=True, index=True)
120    email: Mapped[Optional[str]] = mapped_column(nullable=True)
121    created_at: Mapped[datetime] = mapped_column(default=datetime.utcnow)
122    
123    # Relationships
124    downloads: Mapped[list["Download"]] = relationship(
125        back_populates="user",
126        lazy="selectin"
127    )
128```
129 
130## Migrations (Alembic)
131 
132### Creating Migrations
133- Use Alembic for schema migrations
134- Configuration in `alembic.ini`
135- Migration scripts in `migrations/versions/`
136- Use descriptive migration messages
137 
138### Migration Commands
139```bash
140# Create new migration
141alembic revision --autogenerate -m "Add user table"
142 
143# Apply migrations
144alembic upgrade head
145 
146# Rollback migration
147alembic downgrade -1
148```
149 
150### Migration Best Practices
151- Review auto-generated migrations before applying
152- Test migrations in development first
153- Include both upgrade and downgrade paths
154- Add data migrations when needed
155- Never modify applied migrations
156 
157## Database Configuration
158 
159### Connection Settings
160- Configure database URL in `app/core/config.py`
161- Use async SQLite driver: `aiosqlite`
162- Connection string format: `sqlite+aiosqlite:///./data/hermes.db`
163 
164### Session Configuration
165- Pool settings configured in `app/db/session.py`
166- Echo SQL in development (for debugging)
167- Disable echo in production
168 
169## Transactions
170 
171### Transaction Boundaries
172- FastAPI handles transactions automatically
173- Manual transaction control when needed:
174 
175```python
176async with session.begin():
177    # Multiple operations in one transaction
178    user = await repos["users"].create(user_data)
179    await repos["api_keys"].create(api_key_data)
180```
181 
182### Error Handling
183- SQLAlchemy exceptions should be caught and converted to HTTP exceptions
184- Rollback happens automatically on exceptions
185- Log database errors with context
186 
187## Testing
188 
189### Test Database
190- Use separate database for tests
191- Reset database state between tests
192- Use fixtures for common test data
193 
194### Repository Testing
195- Test each repository method independently
196- Mock external dependencies
197- Verify database state after operations
198- Test error conditions
199