(
session: SessionDep,
current_user: CurrentUser,
collection_in: CollectionCreate,
provider: GeminiProviderDep,
)
| 43 | |
| 44 | @router.post("/collections/", response_model=Collection) |
| 45 | async def create_collection( |
| 46 | session: SessionDep, |
| 47 | current_user: CurrentUser, |
| 48 | collection_in: CollectionCreate, |
| 49 | provider: GeminiProviderDep, |
| 50 | ) -> Any: |
| 51 | name = collection_in.name |
| 52 | cards = None |
| 53 | |
| 54 | if collection_in.prompt: |
| 55 | try: |
| 56 | if not await asyncio.to_thread( |
| 57 | lambda: check_and_increment_ai_usage_quota(session, current_user) |
| 58 | ): |
| 59 | raise HTTPException( |
| 60 | status_code=429, detail="Quota for AI usage is reached." |
| 61 | ) |
| 62 | flashcard_collection = await services.generate_ai_collection( |
| 63 | provider, collection_in.prompt |
| 64 | ) |
| 65 | name = flashcard_collection.name |
| 66 | cards = flashcard_collection.cards |
| 67 | except EmptyCollectionError: |
| 68 | raise HTTPException( |
| 69 | status_code=400, detail="Failed to generate flashcards from the prompt" |
| 70 | ) |
| 71 | except AIGenerationError as e: |
| 72 | raise HTTPException(status_code=500, detail=str(e)) |
| 73 | |
| 74 | return await asyncio.to_thread( |
| 75 | lambda: services.create_collection( |
| 76 | session=session, user_id=current_user.id, name=name, cards=cards |
| 77 | ) |
| 78 | ) |
| 79 | |
| 80 | |
| 81 | @router.get("/collections/{collection_id}", response_model=Collection) |
nothing calls this directly
no test coverage detected