initial commit

Author: Cinex10

.github/workflows/mlops_workflow.yml

@@ -0,0 +1,23 @@
+name: MLOps Workflow
+
+on:
+  push:
+    branches: [ main ]
+  pull_request:
+    branches: [ main ]
+
+jobs:
+  test:
+    runs-on: ubuntu-latest
+    steps:
+    - uses: actions/checkout@v2
+    - name: Set up Python
+      uses: actions/setup-python@v2
+      with:
+        python-version: '3.9'
+    - name: Install dependencies
+      run: |
+        python -m pip install --upgrade pip
+        pip install -r requirements.txt
+    - name: Run tests
+      run: pytest

.gitignore

@@ -0,0 +1,2 @@
+.env
+env

README.md

@@ -0,0 +1 @@
+# Codinsight Fast Api Server

requirements.txt

@@ -0,0 +1,9 @@
+fastapi
+pydantic[email]
+requests
+pydantic-settings
+pymongo
+pytest
+httpx
+groq
+pytest-asyncio

src/__init__.py

@@ -0,0 +1,51 @@
+from pydantic import BaseModel, BeforeValidator, Field, EmailStr, GetJsonSchemaHandler, ConfigDict
+from typing import Any, Callable,  Optional, Annotated
+from datetime import datetime
+from bson import ObjectId
+from pydantic_core import core_schema
+
+
+
+PyObjectId = Annotated[str, BeforeValidator(str)]
+
+
+class UserModel(BaseModel):
+    model_config = ConfigDict(
+        populate_by_name=True,
+        arbitrary_types_allowed = True,
+        json_encoders = {ObjectId: str}
+        )
+    
+    id: Optional[PyObjectId] = Field(alias="_id", default=None)
+    email: EmailStr = Field(...)
+    password: str = Field(...)
+
+class CompletionModel(BaseModel):
+    model_config = ConfigDict(
+        populate_by_name=True,
+        arbitrary_types_allowed = True,
+        json_encoders = {ObjectId: str}
+        )
+    
+    id: Optional[PyObjectId] = Field(alias="_id", default=None)
+    prompt: str = Field(...)
+    completion: str = Field(...)
+    inference_duration_in_ms: float = Field(gt=0)
+    user_ip: str = Field(...)
+    user_device: str = Field(...)
+    user_id: Optional[PyObjectId] = Field(None)
+    timestamp: datetime = Field(default_factory=lambda: datetime.now())
+    rating: Optional[int] = Field(None, ge=1, le=5)
+    
+    def __eq__(self, other):
+        if isinstance(other, CompletionModel):
+            return (
+                self.prompt == other.prompt and
+                self.completion == other.completion and
+                self.user_ip == other.user_ip and
+                self.user_device == other.user_device and
+                self.user_id == other.user_id and
+                self.rating == other.rating and
+                self.inference_duration_in_ms == other.inference_duration_in_ms
+            )
+        return False

src/code_explainer/__init__.py

@@ -0,0 +1,25 @@
+from pydantic import BaseModel, EmailStr, Field
+from typing import Optional, List
+from datetime import datetime
+
+# Completion Schemas
+class CompletionCreate(BaseModel):
+    prompt: str = Field(..., min_length=1)
+    user_ip: Optional[str] = None
+    user_device: Optional[str] = None
+
+class CompletionResponse(BaseModel):
+    id: str
+    prompt: str
+    completion: str
+
+class CompletionUpdate(BaseModel):
+    rating: int = Field(..., ge=1, le=5)
+
+class CompletionList(BaseModel):
+    total: int
+    completions: List[CompletionResponse]
+
+# Error Schemas
+class HTTPError(BaseModel):
+    detail: str

src/code_explainer/constants.py

@@ -0,0 +1,58 @@
+import time
+from typing import Dict, Any
+from src.code_explainer.models import CompletionModel
+from src.config import settings
+from groq import Groq
+
+from src.database import get_db
+from src.exceptions import ExternalServiceError
+
+class CodeExplainService:
+    def __init__(self):
+        self.api_key = settings.LLM_API_KEY
+        self.model = settings.MODEL
+        self.db = get_db()
+        
+    
+    def save_completion(self, completion: CompletionModel):
+        completion_dict = completion.model_dump(by_alias=True)
+        self.db.completions.insert_one(completion_dict)
+
+    async def get_code_explanation(self, code: str, user_ip: str, user_device: str) -> Dict[str, Any]:
+        start_time = time.time()
+        
+        client = Groq(api_key= self.api_key)
+
+        try:
+            chat_completion = client.chat.completions.create(
+                messages=
+                [
+                    {
+                        "role": "user",
+                        "content": code
+                    }
+                ],
+                model=self.model,)
+        except:
+            raise ExternalServiceError(service='groq', detail='')
+
+        end_time = time.time()
+        duration = end_time - start_time
+        
+        explanation = chat_completion.choices[0].message.content
+
+        completion = CompletionModel(
+            prompt=code,
+            completion=explanation,
+            user_ip= user_ip,
+            user_device=user_device,
+            inference_duration_in_ms=duration,
+            user_id= None,
+            rating=None
+        )
+        
+        self.save_completion(completion)
+        
+        return explanation, duration
+
+code_explain_service = CodeExplainService()

src/code_explainer/exceptions.py

@@ -0,0 +1,33 @@
+from pydantic_settings import BaseSettings, SettingsConfigDict
+from functools import lru_cache
+
+class Settings(BaseSettings):
+    model_config = SettingsConfigDict(env_file='.env', env_file_encoding='utf-8')
+
+    # Database settings
+    DATABASE_URL: str = "sqlite:///./test.db"
+    
+    # API settings
+    API_V1_STR: str = "/api/v1"
+    PROJECT_NAME: str = "Codinsight"
+    
+    # Security settings
+    SECRET_KEY: str = "YOUR_SECRET_KEY_HERE"
+    ALGORITHM: str = "HS256"
+    ACCESS_TOKEN_EXPIRE_MINUTES: int = 30
+    
+    # AWS settings
+    AWS_ACCESS_KEY_ID: str = "YOUR_AWS_ACCESS_KEY_ID"
+    AWS_SECRET_ACCESS_KEY: str = "YOUR_AWS_SECRET_ACCESS_KEY"
+    AWS_REGION: str = "us-west-2"
+
+    LLM_API_KEY: str = "YOUR_LLM_API_KEY"
+    MODEL: str = "YOUR_MODEL"
+    
+@lru_cache()
+def get_settings():
+    return Settings()
+
+# Usage
+settings = get_settings()
+print(settings.DATABASE_URL)

src/code_explainer/models.py

@@ -0,0 +1,13 @@
+from pymongo import MongoClient
+from pymongo.database import Database
+from src.config import settings
+
+client = MongoClient(settings.DATABASE_URL)
+db = client.get_database(name='alpha')
+
+def get_db() -> Database:
+    return db
+
+# Optional: Function to close the connection
+def close_mongo_connection():
+    client.close()

src/code_explainer/router.py

@@ -0,0 +1,34 @@
+from fastapi import HTTPException, status
+
+class NotFoundError(HTTPException):
+    def __init__(self, detail: str):
+        super().__init__(status_code=status.HTTP_404_NOT_FOUND, detail=detail)
+
+class UnauthorizedError(HTTPException):
+    def __init__(self, detail: str):
+        super().__init__(
+            status_code=status.HTTP_401_UNAUTHORIZED,
+            detail=detail,
+            headers={"WWW-Authenticate": "Bearer"},
+        )
+
+class ForbiddenError(HTTPException):
+    def __init__(self, detail: str):
+        super().__init__(status_code=status.HTTP_403_FORBIDDEN, detail=detail)
+
+class BadRequestError(HTTPException):
+    def __init__(self, detail: str):
+        super().__init__(status_code=status.HTTP_400_BAD_REQUEST, detail=detail)
+
+class InternalServerError(HTTPException):
+    def __init__(self, detail: str = "Internal server error"):
+        super().__init__(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=detail)
+
+class DatabaseError(Exception):
+    def __init__(self, detail: str):
+        self.detail = detail
+
+class ExternalServiceError(Exception):
+    def __init__(self, service: str, detail: str):
+        self.service = service
+        self.detail = detail

src/code_explainer/schemas.py

@@ -0,0 +1,40 @@
+import pytest
+from src.code_explainer.models import CompletionModel
+from src.code_explainer.service import code_explain_service
+    
+@pytest.mark.asyncio
+async def test_get_code_explanation(test_db):
+    
+    # Code example to test 
+    code = '''
+    def add(x,y):
+        return x + y
+    ''';
+    
+    # Mock user data
+    user_ip = '8.8.8.8'
+    user_device = 'mac m1 pro'
+    
+    explanation, duration = await code_explain_service.get_code_explanation(code=code, user_ip=user_ip, user_device=user_device);
+    
+    completion = CompletionModel(
+        prompt=code,
+        completion=explanation,
+        user_ip= user_ip,
+        user_device=user_device,
+        inference_duration_in_ms=duration,
+        user_id= None,
+        rating=None
+    )
+    
+    # Get the last inserted data
+    cursor = test_db.find().sort({'_id': -1}).limit(1)
+    
+    completion2 = [CompletionModel(**data) for data in cursor][0]
+    
+    assert completion2 == completion
+    
+    
+    
+    
+    

src/code_explainer/service.py

@@ -0,0 +1,30 @@
+import pytest
+from src.database import get_db, close_mongo_connection
+
+@pytest.fixture(scope="session", autouse=True)
+def setup_mongo_connection():
+    """
+    Set up the MongoDB connection at the start of the test session.
+    Close the connection after all tests have run.
+    """
+    get_db()  # Initialize the connection
+
+    yield
+
+    close_mongo_connection()  # Close the connection after the session
+
+@pytest.fixture(scope="module")
+def test_db():
+    """
+    Use a test collection within the test database.
+    This fixture is scoped to the module and ensures a clean environment for each test module.
+    """
+    db = get_db()
+    
+    # Create a test collection
+    test_collection = db.completions
+    
+    yield test_collection
+    
+    # Clean up after all tests in the module by dropping the test collection
+    test_collection.drop()

src/code_explainer/utils.py

@@ -0,0 +1,45 @@
+from src.database import get_db
+
+def test_database_connection():
+    db = get_db()
+    assert db is not None, "Database connection failed"
+
+def test_insert_and_retrieve(test_db):
+    # Insert a test document
+    test_document = {"name": "Test User", "email": "test@example.com"}
+    result = test_db.insert_one(test_document)
+    assert result.inserted_id is not None, "Failed to insert test document"
+
+    # Retrieve the test document
+    retrieved_document = test_db.find_one({"_id": result.inserted_id})
+    assert retrieved_document is not None, "Failed to retrieve test document"
+    assert retrieved_document["name"] == "Test User", "Retrieved document data mismatch"
+
+def test_update(test_db):
+    # Insert a test document
+    test_document = {"name": "Update Test", "status": "pending"}
+    result = test_db.insert_one(test_document)
+
+    # Update the document
+    update_result = test_db.update_one(
+        {"_id": result.inserted_id},
+        {"$set": {"status": "completed"}}
+    )
+    assert update_result.modified_count == 1, "Failed to update test document"
+
+    # Verify the update
+    updated_document = test_db.find_one({"_id": result.inserted_id})
+    assert updated_document["status"] == "completed", "Document update failed"
+
+def test_delete(test_db):
+    # Insert a test document
+    test_document = {"name": "Delete Test"}
+    result = test_db.insert_one(test_document)
+
+    # Delete the document
+    delete_result = test_db.delete_one({"_id": result.inserted_id})
+    assert delete_result.deleted_count == 1, "Failed to delete test document"
+
+    # Verify the deletion
+    deleted_document = test_db.find_one({"_id": result.inserted_id})
+    assert deleted_document is None, "Document was not deleted"