chores_lint_errors_and_path_issues (#40)

Makefile

@@ -42,6 +42,7 @@ start-db:
 # ------------------------ #
 
 format-backend:
+	@isort linguaphoto
 	@black linguaphoto
 	@ruff format linguaphoto
 .PHONY: format

apprunner.yaml

@@ -12,7 +12,7 @@ build:
 
 # Specify the start phase command
 run:
-  command: venv/bin/python3 main.py  # Adjust to the path to your application entry point
+  command: venv/bin/python3 linguaphoto/main.py  # Adjust to the path to your application entry point
   env:
     - name: DYNAMODB_TABLE_NAME
       value: "linguaphoto"

debug.py

@@ -0,0 +1,6 @@
+"""it is entity for debugging"""
+import uvicorn
+
+from linguaphoto.main import app
+
+uvicorn.run(app, port=8080, host="0.0.0.0")

linguaphoto/ai/cli.py

@@ -3,6 +3,7 @@
 import argparse
 import asyncio
 import logging
+from io import BytesIO
 from pathlib import Path
 
 from openai import AsyncOpenAI
@@ -29,7 +30,13 @@ async def main() -> None:
     client = AsyncOpenAI(
         api_key="sk-svcacct-PFETCFHtqmHOmIpP_IAyQfBGz5LOpvC6Zudj7d5Wcdp9WjJT4ImAxuotGcpyT3BlbkFJRbtswQqIxYHam9TN13mCM04_OTZE-v8z-Rw1WEcwzyZqW_GcK0PNNyFp6BcA"
     )
-    transcription_response = await transcribe_image(image, client)
+    # Convert the ImageFile to BytesIO
+    image_bytes = BytesIO()
+    image.save(image_bytes, format="JPEG")  # Use the appropriate format for your image
+    image_bytes.seek(0)  # Reset the stream position to the beginning
+
+    # Now call the transcribe_image function with the BytesIO object
+    transcription_response = await transcribe_image(image_bytes, client)
     print(transcription_response.model_dump_json(indent=2))
     with open(root_dir / "transcription.json", "w") as file:
         file.write(transcription_response.model_dump_json(indent=2))

linguaphoto/api/collection.py

@@ -7,7 +7,10 @@
 from linguaphoto.crud.collection import CollectionCrud
 from linguaphoto.errors import NotAuthorizedError
 from linguaphoto.models import Collection
-from linguaphoto.schemas.collection import CollectionCreateFragment, CollectionEditFragment
+from linguaphoto.schemas.collection import (
+    CollectionCreateFragment,
+    CollectionEditFragment,
+)
 from linguaphoto.utils.auth import get_current_user_id
 
 router = APIRouter()
@@ -33,10 +36,11 @@ async def create(
 @router.get("/get_collection", response_model=Collection)
 async def getcollection(
     id: str, user_id: str = Depends(get_current_user_id), collection_crud: CollectionCrud = Depends()
-) -> dict | None:
+) -> Collection:
     async with collection_crud:
         collection = await collection_crud.get_collection(id)
-        print(collection)
+        if collection is None:
+            raise ValueError
         if collection.user != user_id:
             raise NotAuthorizedError
         return collection

linguaphoto/api/image.py

@@ -54,8 +54,9 @@ async def delete_image(
         if image:
             async with collection_crud:
                 collection = await collection_crud.get_collection(image.collection)
-                updated_images = list(filter(lambda image: image != id, collection.images))
-                await collection_crud.edit_collection(image.collection, {"images": updated_images})
+                if collection:
+                    updated_images = list(filter(lambda image: image != id, collection.images))
+                    await collection_crud.edit_collection(image.collection, {"images": updated_images})
                 await image_crud.delete_image(id)
                 return
     raise HTTPException(status_code=400, detail="Image is invalid")

linguaphoto/api/user.py

@@ -10,7 +10,11 @@
     UserSigninRespondFragment,
     UserSignupFragment,
 )
-from linguaphoto.utils.auth import create_access_token, decode_access_token, oauth2_schema
+from linguaphoto.utils.auth import (
+    create_access_token,
+    decode_access_token,
+    oauth2_schema,
+)
 
 router = APIRouter()
 

linguaphoto/crud/base.py

@@ -170,7 +170,7 @@ async def _update_item(
                 raise ValueError(f"Invalid update: {str(e)}")
             raise
 
-    async def _delete_item(self, item: BaseModel | str) -> None:
+    async def _delete_item(self, item: LinguaBaseModel | str) -> None:
         table = await self.db.Table(TABLE_NAME)
         await table.delete_item(Key={"id": item if isinstance(item, str) else item.id})
 

linguaphoto/crud/collection.py

@@ -12,7 +12,7 @@ async def create_collection(self, user_id: str, title: str, description: str) ->
         await self._add_item(collection)
         return collection
 
-    async def get_collection(self, collection_id: str) -> Collection:
+    async def get_collection(self, collection_id: str) -> Collection | None:
         collection = await self._get_item(collection_id, Collection, True)
         return collection
 

linguaphoto/crud/image.py

@@ -73,7 +73,7 @@ async def get_images(self, collection_id: str, user_id: str) -> List[Image]:
         images = await self._get_items_from_secondary_index("user", user_id, Image, Key("collection").eq(collection_id))
         return images
 
-    async def get_image(self, image_id: str) -> Image:
+    async def get_image(self, image_id: str) -> Image | None:
         image = await self._get_item(image_id, Image, True)
         return image
 
@@ -86,6 +86,8 @@ async def translate(self, images: List[str], user_id: str) -> List[Image]:
         for id in images:
             # Retrieve image metadata and download the image content
             image_instance = await self._get_item(id, Image, True)
+            if image_instance is None:
+                continue
             response = requests.get(image_instance.image_url)
             if response.status_code == 200:
                 img_source = BytesIO(response.content)
@@ -102,6 +104,8 @@ async def translate(self, images: List[str], user_id: str) -> List[Image]:
                     # Set buffer position to the start
                     audio_buffer.seek(0)
                     audio_url = await self.create_audio(audio_buffer)
+                    if audio_url is None:
+                        continue
                     # Attach the audio URL to the transcription
                     transcription.audio_url = audio_url
                 image_instance.transcriptions = transcription_response.transcriptions

linguaphoto/crud/user.py

@@ -32,6 +32,5 @@ async def verify_user_by_email(self, user: UserSigninFragment) -> bool:
         else:
             raise ValueError
 
-    async def update_user(self, id: str, data: dict) -> User | None:
-        user = await self._update_item(id, User, data)
-        return user
+    async def update_user(self, id: str, data: dict) -> None:
+        await self._update_item(id, User, data)

linguaphoto/main.py

@@ -5,7 +5,6 @@
 from fastapi.middleware.cors import CORSMiddleware
 
 from linguaphoto.api.api import router
-from linguaphoto.settings import settings
 
 app = FastAPI()
 
@@ -18,94 +17,8 @@
     allow_headers=["*"],
 )
 
-# Retrieve AWS configuration from environment variables
-bucket_name = settings.bucket_name
-dynamodb_table_name = settings.dynamodb_table_name
-media_hosting_server = settings.media_hosting_server
-key_pair_id = settings.key_pair_id
-
 app.include_router(router, prefix="")
 
-
-# class ImageMetadata(BaseModel):
-#     filename: str
-#     s3_url: str
-
-
-# @app.post("/upload/", response_model=ImageMetadata)
-# async def upload_image(file: UploadFile = File(...)) -> ImageMetadata:
-#     if file.filename is None or not file.filename:
-#         raise HTTPException(status_code=400, detail="File name is missing.")
-
-#     try:
-#         # Generate a unique file name
-#         file_extension = file.filename.split(".")[-1] if "." in file.filename else "unknown"
-#         unique_filename = f"{uuid.uuid4()}.{file_extension}"
-
-#         if bucket_name is None:
-#             raise HTTPException(status_code=500, detail="Bucket name is not set.")
-
-#         if dynamodb_table_name is None:
-#             raise HTTPException(status_code=500, detail="DynamoDB table name is not set.")
-
-#         # Create an instance of CloudFrontUrlSigner
-#         private_key_path = os.path.abspath("private_key.pem")
-#         cfs = CloudFrontUrlSigner(str(key_pair_id), private_key_path)
-#         # Generate a signed URL
-#         url = f"{media_hosting_server}/{unique_filename}"
-#         custom_policy = cfs.create_custom_policy(url, expire_days=100)
-#         s3_url = cfs.generate_presigned_url(url, custom_policy)
-#         print(s3_url)
-#         # Create an S3 client with aioboto3
-#         async with aioboto3.Session().client(
-#             "s3",
-#             region_name=settings.aws_region_name,
-#             aws_access_key_id=settings.aws_access_key_id,
-#             aws_secret_access_key=settings.aws_secret_access_key,
-#         ) as s3_client:
-#             # Upload the file to S3
-#             await s3_client.upload_fileobj(file.file, bucket_name, f"uploads/{unique_filename}")
-
-#         # Create a DynamoDB resource with aioboto3
-#         async with aioboto3.Session().resource(
-#             "dynamodb",
-#             region_name=settings.aws_region_name,
-#             aws_access_key_id=settings.aws_access_key_id,
-#             aws_secret_access_key=settings.aws_secret_access_key,
-#         ) as dynamodb:
-#             table = await dynamodb.Table(dynamodb_table_name)
-#             # Save metadata to DynamoDB
-#             await table.put_item(Item={"id": unique_filename, "s3_url": s3_url})
-
-#         return ImageMetadata(filename=unique_filename, s3_url=s3_url)
-
-#     except Exception as e:
-#         print(str(e))
-#         raise HTTPException(status_code=500, detail=str(e))
-
-
-# @app.get("/download/{filename}")
-# async def download_image(filename: str):
-#     try:
-#         async with aioboto3.Session().resource(
-#             "dynamodb",
-#             region_name=settings.aws_region_name,
-#             aws_access_key_id=settings.aws_access_key_id,
-#             aws_secret_access_key=settings.aws_secret_access_key,
-#         ) as dynamodb:
-#             table = await dynamodb.Table(dynamodb_table_name)
-#             # Retrieve image metadata from DynamoDB
-#             response = await table.get_item(Key={"id": filename})
-
-#             if "Item" not in response:
-#                 raise HTTPException(status_code=404, detail="Image not found")
-
-#             # Return the S3 URL for download
-#             return {"s3_url": response["Item"]["s3_url"]}
-
-#     except Exception as e:
-#         raise HTTPException(status_code=500, detail=str(e))
-
 if __name__ == "__main__":
     print("Starting webserver...")
     uvicorn.run(app, port=8080, host="0.0.0.0")

linguaphoto/models.py

@@ -79,7 +79,7 @@ class TranscriptionResponse(BaseModel):
 class Image(LinguaBaseModel):
     is_translated: bool = False
     transcriptions: list[Transcription] = []
-    collection: str | None = None
+    collection: str
     image_url: str
     user: str
 

linguaphoto/settings.py

@@ -18,7 +18,7 @@
 
 
 class Settings:
-    bucket_name = os.getenv("S3_BUCKET_NAME")
+    bucket_name = os.getenv("S3_BUCKET_NAME", "linguaphoto")
     dynamodb_table_name = os.getenv("DYNAMODB_TABLE_NAME", "linguaphoto")
     media_hosting_server = os.getenv("MEDIA_HOSTING_SERVER")
     key_pair_id = os.getenv("KEY_PAIR_ID")
@@ -27,7 +27,7 @@ class Settings:
     aws_secret_access_key = os.getenv("AWS_SECRET_ACCESS_KEY")
     openai_key = os.getenv("OPENAI_API_KEY")
     stripe_key = os.getenv("STRIPE_API_KEY")
-    stripe_price_id = os.getenv("STRIPE_PRODUCT_PRICE_ID")
+    stripe_price_id = os.getenv("STRIPE_PRODUCT_PRICE_ID", "price_1Q0ZaMKeTo38dsfeSWRDGCEf")
 
 
 settings = Settings()

linguaphoto/utils/auth.py

@@ -61,6 +61,8 @@ async def subscription_validate(token: str = Depends(oauth2_schema), user_crud:
         raise HTTPException(status_code=422, detail="Could not validate credentials")
     async with user_crud:
         user = await user_crud.get_user(user_id, True)
+    if user is None:
+        raise HTTPException(status_code=422, detail="Could not validate credentials")
     if user.is_subscription is False:
         raise HTTPException(status_code=422, detail="You need to subscribe.")
     return True

linguaphoto/utils/cloudfront_url_signer.py

@@ -33,7 +33,7 @@ def _rsa_signer(self, message: str) -> bytes:
         with open(self.private_key_path, "r") as key_file:
             private_key = key_file.read()
         return rsa.sign(
-            message,  # Ensure message is in bytes
+            message.encode("utf8"),  # Ensure message is in bytes
             rsa.PrivateKey.load_pkcs1(private_key.encode("utf8")),
             "SHA-1",  # CloudFront requires SHA-1 hash
         )