50 lines
1.3 KiB
Python
50 lines
1.3 KiB
Python
from prisma import Prisma
|
|
from contextlib import asynccontextmanager
|
|
from typing import AsyncGenerator, Optional
|
|
|
|
# Singleton pattern for Prisma client
|
|
_prisma_client: Optional[Prisma] = None
|
|
|
|
async def get_prisma_client() -> Prisma:
|
|
"""
|
|
Get or initialize the Prisma client singleton.
|
|
|
|
Returns:
|
|
Prisma: The initialized Prisma client instance
|
|
"""
|
|
global _prisma_client
|
|
|
|
if _prisma_client is None:
|
|
_prisma_client = Prisma()
|
|
await _prisma_client.connect()
|
|
|
|
return _prisma_client
|
|
|
|
@asynccontextmanager
|
|
async def prisma_client() -> AsyncGenerator[Prisma, None]:
|
|
"""
|
|
Context manager for Prisma client operations.
|
|
|
|
Yields:
|
|
Prisma: The initialized Prisma client instance
|
|
|
|
Example:
|
|
```python
|
|
async with prisma_client() as db:
|
|
users = await db.user.find_many()
|
|
```
|
|
"""
|
|
client = await get_prisma_client()
|
|
try:
|
|
yield client
|
|
except Exception as e:
|
|
print(f"Database operation error: {e}")
|
|
raise
|
|
|
|
async def disconnect_prisma():
|
|
"""Disconnect the Prisma client when shutting down the application."""
|
|
global _prisma_client
|
|
if _prisma_client is not None:
|
|
await _prisma_client.disconnect()
|
|
_prisma_client = None
|