first task

.gitignore

@@ -0,0 +1,7 @@
+.venv/
+.idea/
+
+*.pyc
+*.pyo
+*.pyd
+__pycache__/

hello_world/.env

@@ -0,0 +1,5 @@
+MONGO_INITDB_ROOT_USERNAME=admin
+MONGO_INITDB_ROOT_PASSWORD=1234
+MONGO_HOST=mongodb
+MONGO_DB_NAME=mydb
+MONGO_PORT=27017

hello_world/Dockerfile

@@ -0,0 +1,11 @@
+FROM python:3.9
+
+WORKDIR /app
+
+COPY requirements.txt .
+
+RUN pip install --no-cache-dir -r requirements.txt
+
+COPY . .
+
+CMD ["uvicorn", "src.main:app", "--host", "0.0.0.0", "--port", "8080"]

hello_world/docker-compose.yml

@@ -0,0 +1,20 @@
+version: '3.8'
+
+services:
+  mongo:
+    image: mongo:7.0
+    hostname: mongodb
+    env_file:
+      .env
+    volumes:
+      - mongo-data:/data/db
+
+  app:
+    build: .
+    ports:
+      - "127.0.0.1:8080:8080"
+    depends_on:
+      - mongo
+
+volumes:
+  mongo-data:

hello_world/requirements.txt

@@ -0,0 +1,16 @@
+annotated-types==0.7.0
+anyio==4.6.0
+click==8.1.7
+dnspython==2.6.1
+fastapi==0.115.0
+h11==0.14.0
+idna==3.10
+motor==3.6.0
+pydantic==2.9.2
+pydantic_core==2.23.4
+pymongo==4.9.1
+python-dotenv==1.0.1
+sniffio==1.3.1
+starlette==0.38.5
+typing_extensions==4.12.2
+uvicorn==0.30.6

hello_world/src/api/text_router.py

@@ -0,0 +1,52 @@
+from fastapi import APIRouter, FastAPI, Request, HTTPException
+from fastapi.params import Depends
+from src.reposiroties.text_repository import TextRepository
+from src.schemas.text import TextInDB, Text, TextUpdate
+from src.services.text_service import TextService
+
+router = APIRouter(
+    prefix="/texts",
+    tags=["Text"]
+)
+
+@router.get("/", response_model=list[TextInDB])
+async def get_texts(request: Request) -> list[TextInDB]:
+    text_service: TextService = request.app.text_service
+    result = await text_service.get_all_texts()
+    return result
+
+@router.post("/", response_model=TextInDB)
+async def create_text(request: Request, text: Text) -> TextInDB:
+    text_service: TextService = request.app.text_service
+    result = await text_service.create_text(text)
+    return result
+
+@router.get("/{text_id}", response_model=TextInDB)
+async def get_text_by_id(request: Request, text_id: str) -> TextInDB:
+    text_service: TextService = request.app.text_service
+    result = await text_service.get_text_by_id(text_id)
+
+    if not result:
+        raise HTTPException(status_code=404, detail="Text not found")
+
+    return result
+
+@router.delete("/{text_id}", response_model=TextInDB)
+async def delete_text(request: Request, text_id: str) -> TextInDB:
+    text_service: TextService = request.app.text_service
+    result = await text_service.delete_text(text_id)
+
+    if not result:
+        raise HTTPException(status_code=404, detail="Text not found")
+
+    return result
+
+@router.put("/", response_model=TextInDB)
+async def update_text(request: Request, text: TextUpdate) -> TextInDB:
+    text_service: TextService = request.app.text_service
+    result = await text_service.update_text(text)
+
+    if not result:
+        raise HTTPException(status_code=404, detail="Text not found")
+
+    return result

hello_world/src/config.py

@@ -0,0 +1,12 @@
+import os
+from dotenv import load_dotenv
+
+load_dotenv()
+
+MONGO_HOST = os.getenv("MONGO_HOST")
+MONGO_USER = os.getenv("MONGO_INITDB_ROOT_USERNAME")
+MONGO_PASS = os.getenv("MONGO_INITDB_ROOT_PASSWORD")
+MONGO_PORT = os.getenv("MONGO_PORT")
+MONGO_DB_NAME = os.getenv("MONGO_DB_NAME")
+
+MONGO_URI = f"mongodb://{MONGO_USER}:{MONGO_PASS}@{MONGO_HOST}:{MONGO_PORT}/{MONGO_DB_NAME}?authSource=admin"

hello_world/src/database/database.py

@@ -0,0 +1,12 @@
+from motor.motor_asyncio import AsyncIOMotorClient, AsyncIOMotorDatabase
+
+
+async def connect_to_database(uri: str) -> tuple[AsyncIOMotorClient, AsyncIOMotorDatabase]:
+    client = AsyncIOMotorClient(uri)
+    db = client.get_default_database()
+    pong = await db.command("ping")
+
+    if pong.get("ok") != 1:
+        raise Exception("Could not connect to database")
+
+    return client, db

hello_world/src/main.py

@@ -0,0 +1,22 @@
+from contextlib import asynccontextmanager
+from fastapi import FastAPI
+from src.config import MONGO_URI
+from src.database.database import connect_to_database
+from src.reposiroties.mongo_text_repository import MongoTextRepository
+from src.api.text_router import router
+from src.services.text_service import TextService
+
+@asynccontextmanager
+async def lifespan(app: FastAPI):
+    app.mongo_client, app.mongodb = await connect_to_database(MONGO_URI)
+    repository = MongoTextRepository(app.mongodb)
+    app.text_service = TextService(repository)
+    yield
+
+    if app.mongo_client:
+        app.mongo_client.close()
+
+app = FastAPI(lifespan=lifespan)
+
+app.include_router(router)
+

hello_world/src/reposiroties/mongo_text_repository.py

@@ -0,0 +1,48 @@
+from typing import Optional
+from bson import ObjectId
+from src.reposiroties.text_repository import TextRepository
+from motor.motor_asyncio import AsyncIOMotorDatabase
+from src.schemas.text import Text, TextInDB, TextUpdate
+
+
+class MongoTextRepository(TextRepository):
+    def __init__(self, db: AsyncIOMotorDatabase):
+        self.db = db
+
+    async def create_text(self, text: Text) -> TextInDB:
+        text_data = text.model_dump()
+        result = await self.db["texts"].insert_one(text_data)
+        text_data["id"] = str(result.inserted_id)
+        return TextInDB(**text_data)
+
+    async def get_text_by_id(self, text_id: str) -> Optional[TextInDB]:
+        result = await self.db["texts"].find_one({"_id": ObjectId(text_id)})
+
+        if result:
+            return TextInDB.from_mongo(result)
+
+        return None
+
+    async def get_all_texts(self) -> list[TextInDB]:
+        cursor = self.db["texts"].find()
+        return [TextInDB.from_mongo(doc) async for doc in cursor]
+
+    async def update_text(self, text: TextUpdate) -> Optional[TextInDB]:
+        update_data = {"$set": text.model_dump()}
+        result = await self.db["texts"].find_one_and_update(
+            {"_id": ObjectId(text.id)}, update_data, return_document=True
+        )
+
+        if result:
+            return TextInDB.from_mongo(result)
+
+        return None
+
+    async def delete_text(self, text_id: str) -> Optional[TextInDB]:
+        document = await self.db["texts"].find_one({"_id": ObjectId(text_id)})
+
+        if document:
+            await self.db["texts"].delete_one({"_id": ObjectId(text_id)})
+            return TextInDB.from_mongo(document)
+
+        return None

hello_world/src/reposiroties/text_repository.py

@@ -0,0 +1,26 @@
+from abc import ABC, abstractmethod
+from typing import Optional
+
+from src.schemas.text import Text, TextInDB, TextUpdate
+
+
+class TextRepository(ABC):
+    @abstractmethod
+    async def create_text(self, text: Text) -> TextInDB:
+        pass
+
+    @abstractmethod
+    async def get_text_by_id(self, text_id: str) -> Optional[TextInDB]:
+        pass
+
+    @abstractmethod
+    async def get_all_texts(self) -> list[TextInDB]:
+        pass
+
+    @abstractmethod
+    async def update_text(self, text: TextUpdate) -> Optional[TextInDB]:
+        pass
+
+    @abstractmethod
+    async def delete_text(self, text_id: str) -> Optional[TextInDB]:
+        pass

hello_world/src/schemas/text.py

@@ -0,0 +1,18 @@
+from pydantic import BaseModel, Field
+
+
+class Text(BaseModel):
+    text: str
+
+class TextInDB(BaseModel):
+    id: str
+    text: str
+
+    @classmethod
+    def from_mongo(cls, doc):
+        return cls(id=str(doc["_id"]), text=doc["text"])
+
+
+class TextUpdate(BaseModel):
+    id: str
+    text: str

hello_world/src/services/text_service.py

@@ -0,0 +1,24 @@
+from typing import Optional
+
+from src.reposiroties.text_repository import TextRepository
+from src.schemas.text import TextInDB, Text, TextUpdate
+
+
+class TextService:
+    def __init__(self, repository: TextRepository):
+        self.repository = repository
+
+    async def create_text(self, text: Text) -> TextInDB:
+        return await self.repository.create_text(text)
+
+    async def get_text_by_id(self, text_id: str) -> Optional[TextInDB]:
+        return await self.repository.get_text_by_id(text_id)
+
+    async def get_all_texts(self) -> list[TextInDB]:
+        return await self.repository.get_all_texts()
+
+    async def update_text(self, text: TextUpdate) -> Optional[TextInDB]:
+        return await self.repository.update_text(text)
+
+    async def delete_text(self, text_id: str) -> Optional[TextInDB]:
+        return await self.repository.delete_text(text_id)