FastAPI added

Python/FastAPI/Dockerfile

@@ -0,0 +1,22 @@
+# Use official Python image
+FROM python:slim
+
+# Set work directory
+WORKDIR /app
+
+# Update system packages
+RUN apt-get update && apt-get upgrade -y
+
+# Copy dependencies first
+COPY requirements.txt .
+
+# Install Python dependencies
+RUN pip install --upgrade pip && pip install -r requirements.txt
+
+# Copy the rest of your app
+COPY . .
+
+# Run create_models.py before starting the app
+RUN chmod +x wait-for-it.sh
+CMD ["sh", "-c", "./wait-for-it.sh postgres:5432 -- python create_models.py && uvicorn main:app --host 0.0.0.0 --port 8100"]
+# CMD ["/bin/bash"]

Python/FastAPI/create_models.py

@@ -0,0 +1,10 @@
+from models import Base
+from db import engine
+import asyncio
+
+async def create_tables():
+    async with engine.begin() as conn:
+        await conn.run_sync(Base.metadata.create_all)
+
+if __name__ == "__main__":
+    asyncio.run(create_tables())

Python/FastAPI/db.py

@@ -0,0 +1,17 @@
+from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
+from sqlalchemy.orm import sessionmaker
+import asyncio
+
+engine = create_async_engine("postgresql+asyncpg://resttest:resttest@postgres:5432/resttest", echo=True)
+
+AsyncSessionClient = sessionmaker(
+    bind = engine,
+    class_ = AsyncSession
+)
+
+async def get_db():
+    async with AsyncSessionClient() as session:
+        try:
+            yield session
+        finally:
+            await session.close()

Python/FastAPI/dependencies.py

@@ -0,0 +1,8 @@
+from db import get_db, AsyncSession
+from models import User
+from sqlalchemy import select
+from fastapi import Depends
+from repositories import UserRepository
+
+async def get_user_repository(db: AsyncSession = Depends(get_db)) -> UserRepository:
+    return UserRepository(db_session=db)

Python/FastAPI/main.py

@@ -1,28 +1,103 @@
 import asyncio
 import time
-from fastapi import FastAPI
-from hypercorn.asyncio import serve
-from hypercorn.config import Config
+import json
+from fastapi import HTTPException, status
+from fastapi import FastAPI, Path, Body, Depends
+from redis.asyncio import Redis
+from repositories import UserRepository
+from dependencies import get_user_repository
+from db import get_db
+from schemas import UserSchema
+from models import User
+from redis.exceptions import RedisError
+from contextlib import asynccontextmanager
 
-app = FastAPI()
+redis_client: Redis | None = None
 
-@app.get("/test/get-users")
-async def get_users()-> list[dict[str, str | int]]:
+@asynccontextmanager
+async def lifespan(app: FastAPI):
+    global redis_client
+    redis_client = Redis(host="redis", port=6379)
+    try:
+        await redis_client.ping()
+        print("Connected to Redis")
+    except Exception as e:
+        print(f'Redis connection failed: {e}')
+    yield
+    await redis_client.close()
+    await redis_client.connection_pool.disconnect()
+
+app = FastAPI(lifespan=lifespan)
+
+@app.get("/health/")
+async def health():
+    start = time.perf_counter()
+    print(f'Execution time: {(time.perf_counter() - start) *1000:3f}ms')
+    return {
+        "status": "ok"
+    }
+
+@app.get("/user/json/")
+async def user_json():
+    start = time.perf_counter()
+    print(f'Execution time: {(time.perf_counter() - start) *1000:3f}ms')
+    return {
+        "username": "Alice",
+        "email": "alice@example.com"
+    }
+
+@app.get("/user/db/{id}/")
+async def user_db_id(
+    id: int = Path(),
+    user_repository: UserRepository = Depends(get_user_repository)
+):
+    start = time.perf_counter()
+    user = await user_repository.get_by_id(id)
+    if not user:
+        print(f'Execution time: {(time.perf_counter() - start) *1000:3f}ms')
+        raise HTTPException(status=status.HTTP_404_NOT_FOUND, detail="user not found")
+    print(f'Execution time: {(time.perf_counter() - start) *1000:3f}ms')
+    return user
+
+@app.post("/user/db/")
+async def user_db(
+    user: UserSchema = Body(),
+    user_repository: UserRepository = Depends(get_user_repository)
+):
+    start = time.perf_counter()
+    item = await user_repository.create(
+        User(username=user.username, email=user.email)
+    )
+    print(f'Execution time: {(time.perf_counter() - start) *1000:3f}ms')
+    return item
+
+@app.get("/user/cache/{id}/")
+async def user_cache_id(
+    id: int = Path()
+):
     start = time.perf_counter()
-    users = [{
-            "id": user,
-            "username": f"user{user}",
-            "email": f"user{user}@gmail.com",
-            "password": f"password{user}",
-    } for user in range(10000)]
-    print(f"Execution time: {(time.perf_counter() - start) *1000:3f}ms")
-    return users
-
-async def run():
-    config = Config()
-    config.bind = ["0.0.0.0:8100"]
-    await serve(app, config=config)
-
-if __name__ == "__main__":
-    asyncio.run(run())
+    try:
+        cached = await redis_client.get(id)
+        if cached:
+            print(type(cached))
+            data = json.loads(cached.decode('utf-8'))
+            print(type(data))
+            user_schema = UserSchema(username=data.get('username'), email=data.get('email'))
+            print(f'Execution time: {(time.perf_counter() - start) *1000:3f}ms')
+            return user_schema
+        db = await get_db().__anext__()
+        user_repository = await get_user_repository(db)
+        user = await user_repository.get_by_id(id)
+        if not user:
+            print(f'Execution time: {(time.perf_counter() - start) *1000:3f}ms')
+            raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="user not found")
+        user_schema = UserSchema(username=user.username, email=user.email)
+        user_json = user_schema.model_dump_json()
+        await redis_client.set(id, user_json, ex=300)
+        print(f'Execution time: {(time.perf_counter() - start) *1000:3f}ms')
+        return user_schema
+    except RedisError as e:
+        print(f'Redis error: {e}')
+        print(f'Execution time: {(time.perf_counter() - start) *1000:3f}ms')
+        return None
 

Python/FastAPI/models.py

@@ -0,0 +1,13 @@
+from sqlalchemy.orm import DeclarativeBase
+from sqlalchemy.orm import mapped_column, Mapped
+from sqlalchemy import Integer, String
+
+class Base(DeclarativeBase):
+    pass
+
+class User(Base):
+    __tablename__ = "users"
+
+    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
+    username: Mapped[str] = mapped_column(String, unique=True, nullable=False)
+    email: Mapped[str] = mapped_column(String, unique=True, nullable=False)

Python/FastAPI/repositories.py

@@ -0,0 +1,40 @@
+from sqlalchemy.ext.asyncio import AsyncSession
+from models import User
+from sqlalchemy import select
+from fastapi import HTTPException, status
+from sqlalchemy.sql.selectable import Select
+
+class UserRepository:
+    def __init__(self, db_session: AsyncSession):
+        self.db_session = db_session
+
+    async def create(self, item: User) -> User:
+        obj = await self.get_by_username(item.username)
+        if obj is None:
+            obj = await self.get_by_email(item.email)
+            if obj is None:
+                self.db_session.add(item)   
+                await self.db_session.commit()
+                await self.db_session.refresh(item)
+                return item
+            raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="email is not free")
+        print(obj.username, obj.email, obj.id, obj)
+        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="username is not free")
+
+    async def get_by_id(self, id: int) -> User:
+        user = await self.db_session.get(User, id)
+        return user
+
+    async def execute_stmt(self, stmt: Select) -> User | None:
+        result = await self.db_session.execute(stmt)
+        user = result.scalar_one_or_none()
+        return user
+
+    async def get_by_email(self, email: str) -> User | None:
+        stmt = select(User).where(User.email == email)
+        return await self.execute_stmt(stmt)
+
+    async def get_by_username(self, username: str) -> User | None:
+        stmt = select(User).where(User.username == username)
+        return await self.execute_stmt(stmt)
+

Python/FastAPI/requirements.txt

@@ -0,0 +1,43 @@
+annotated-types==0.7.0
+anyio==4.9.0
+asyncpg==0.30.0
+certifi==2025.7.14
+click==8.2.1
+colorama==0.4.6
+dnspython==2.7.0
+email_validator==2.2.0
+fastapi==0.116.1
+fastapi-cli==0.0.8
+fastapi-cloud-cli==0.1.4
+greenlet==3.2.3
+h11==0.16.0
+httpcore==1.0.9
+httptools==0.6.4
+httpx==0.28.1
+idna==3.10
+Jinja2==3.1.6
+markdown-it-py==3.0.0
+MarkupSafe==3.0.2
+mdurl==0.1.2
+pydantic==2.11.7
+pydantic_core==2.33.2
+Pygments==2.19.2
+python-dotenv==1.1.1
+python-multipart==0.0.20
+PyYAML==6.0.2
+redis==6.2.0
+rich==14.0.0
+rich-toolkit==0.14.8
+rignore==0.6.2
+sentry-sdk==2.33.0
+shellingham==1.5.4
+sniffio==1.3.1
+SQLAlchemy==2.0.41
+starlette==0.47.1
+typer==0.16.0
+typing-inspection==0.4.1
+typing_extensions==4.14.1
+urllib3==2.5.0
+uvicorn==0.35.0
+watchfiles==1.1.0
+websockets==15.0.1

Python/FastAPI/schemas.py

@@ -0,0 +1,7 @@
+from pydantic import BaseModel
+class UserSchema(BaseModel):
+    username: str
+    email: str
+
+    class Config:
+        orm_mode = True