Add new files and models

Dockerfile

@@ -0,0 +1,12 @@
+FROM python:3.11.7-bookworm
+
+WORKDIR /app
+
+COPY ./app/ .
+
+RUN pip install --no-cache-dir --upgrade -r /app/requirements.txt
+
+ENV SUPABASE_URL='https://lakewjtjnwmihctodrow.supabase.co'
+ENV SUPABASE_SECRET_KEY='eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJpc3MiOiJzdXBhYmFzZSIsInJlZiI6Imxha2V3anRqbndtaWhjdG9kcm93Iiwicm9sZSI6ImFub24iLCJpYXQiOjE2OTg0MTgwMTMsImV4cCI6MjAxMzk5NDAxM30.6UI7vAzZXZFoSXPaJ2v8LQYpHmzAAuBVykWXV8lMLEA'
+EXPOSE 5000
+CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "7860"]

app/__init__.py

@@ -0,0 +1 @@
+# This file is intentionally left blank

app/crud/user.py

@@ -0,0 +1,25 @@
+from supabase import Client
+from utils.auth import get_user_response
+from models.user import User
+
+
+async def update_user(
+    supabase: Client, email: str = None, password: str = None, data: dict = None
+):
+    info = {}
+    if email:
+        info["email"] = email
+    if password:
+        info["password"] = password
+    if data:
+        info["data"] = data
+    supabase.auth.update_user(info)
+    return True
+
+
+async def get_user(supabase: Client, token: str):
+    user_response = get_user_response(token, supabase)
+    if user_response:
+        return User(email=user_response.user.email, **user_response.user.user_metadata)
+    else:
+        return None

app/db/__init__.py

@@ -0,0 +1 @@
+# This file is intentionally left blank

app/db/supabase_service.py

@@ -0,0 +1,13 @@
+from supabase import create_client
+from dotenv import load_dotenv
+import os
+
+load_dotenv()
+
+SUPABASE_URL = os.getenv("SUPABASE_URL")
+SUPABASE_SECRET_KEY = os.getenv("SUPABASE_SECRET_KEY")
+SUPABASE = create_client(SUPABASE_URL, SUPABASE_SECRET_KEY)
+
+
+def get_supabase():
+    return SUPABASE

app/main.py

@@ -0,0 +1,26 @@
+from fastapi.responses import RedirectResponse
+from fastapi import FastAPI
+from fastapi.middleware.cors import CORSMiddleware
+from routers import authentication, users, group_admin, song, album, concert, playlist
+
+app = FastAPI()
+
+app.include_router(authentication.router)
+app.include_router(users.router)
+app.include_router(group_admin.router)
+app.include_router(song.router)
+app.include_router(album.router)
+app.include_router(concert.router)
+app.include_router(playlist.router)
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=["*"],
+    allow_credentials=True,
+    allow_methods=["*"],
+    allow_headers=["*"],
+)
+
+
+@app.get("/")
+def read_root():
+    return RedirectResponse(url="/docs")

app/models/__init__.py

@@ -0,0 +1 @@
+# This file is intentionally left blank

app/models/concert.py

@@ -0,0 +1,15 @@
+from pydantic import BaseModel
+
+
+class ConcertInfo(BaseModel):
+    name: str
+    location: str
+    description: str | None = None
+    start_at: str
+    number_of_tickets: int = 0
+
+
+class Concert(ConcertInfo):
+    id: int
+    created_at: str
+    group_id: int

app/models/enums.py

@@ -0,0 +1,13 @@
+from enum import Enum
+
+
+class Role(str, Enum):
+    user = "user"
+    admin = "admin"
+    group_admin = "group_admin"
+
+
+class Gender(Enum):
+    male = 0
+    female = 1
+    other = 2

app/models/group.py

@@ -0,0 +1,16 @@
+from pydantic import BaseModel
+
+
+class GroupBase(BaseModel):
+    id: int
+    created_at: str
+
+
+class GroupInfo(BaseModel):
+    name: str
+    description: str | None = None
+
+
+class Group(GroupBase, GroupInfo):
+    admin_id: str
+    members_id: list[str] | None = None

app/models/music.py

@@ -0,0 +1,30 @@
+from pydantic import BaseModel
+
+
+class Song(BaseModel):
+    id: int
+    name: str
+    created_at: str
+    user_id: str
+    name_in_storage: str
+    description: str | None
+    lyric: str | None
+
+
+class BaseAlbum(BaseModel):
+    name: str
+    description: str | None
+
+
+class Album(BaseAlbum):
+    id: int
+    created_at: str
+    user_id: str
+
+class BasePlaylist(BaseModel):
+    name: str
+
+class Playlist(BasePlaylist):
+    id: int
+    created_at: str
+    user_id: str

app/models/user.py

@@ -0,0 +1,28 @@
+from typing import Optional
+from pydantic import BaseModel, Field
+from datetime import date
+from .enums import Gender, Role
+
+
+class UserBase(BaseModel):
+    email: str = Field(description="Email of user", example="[email protected]")
+
+
+class UserLogin(UserBase):
+    password: str
+
+
+class UserInfo(BaseModel):
+    name: str
+    birthdate: date | None = Field(example="2000-01-01")
+    phone_number: str | None = Field(example="0382929292")
+    gender: Gender
+    role: Role = Role.user
+
+
+class UserSignup(UserLogin, UserInfo):
+    pass
+
+
+class User(UserBase, UserInfo):
+    pass

app/routers/__init__.py

@@ -0,0 +1 @@
+# This file is intentionally left blank

app/routers/album.py

@@ -0,0 +1,106 @@
+from fastapi import (
+    APIRouter,
+    Depends,
+    HTTPException,
+    status,
+    Security,
+)
+from models.music import Album, BaseAlbum
+from db.supabase_service import get_supabase
+from typing import Annotated, List, Tuple
+from supabase import Client
+from utils.auth import get_id
+from fastapi.encoders import jsonable_encoder
+from utils.exceptions import BAD_REQUEST
+
+router = APIRouter(tags=["Album"], prefix="/album")
+
+
+@router.post("", description="Create album")
+async def create_album(
+    supabase: Annotated[Client, Depends(get_supabase)],
+    user_id: Annotated[str, Security(get_id)],
+    album: BaseAlbum,
+):
+    try:
+        supabase.table("albums").insert(
+            {"name": album.name, "description": album.description, "user_id": user_id}
+        ).execute()
+        return {"detail": "Album created"}
+    except:
+        raise BAD_REQUEST
+
+
+# delete album
+@router.delete("", description="Delete album")
+async def delete_album(
+    supabase: Annotated[Client, Depends(get_supabase)],
+    user_id: Annotated[str, Security(get_id)],
+    album_id: int,
+):
+    try:
+        supabase.table("albums").delete().eq("id", album_id).execute()
+        return {"detail": "Album deleted"}
+    except:
+        raise BAD_REQUEST
+
+
+# get list of all album
+@router.get("/all", description="Get list of all album", response_model=List[Album])
+async def get_all_album(
+    supabase: Annotated[Client, Depends(get_supabase)],
+    user_id: Annotated[str, Security(get_id)],
+) -> List[Album]:
+    try:
+        res = supabase.table("albums").select("*").execute().dict()["data"]
+        return res
+    except:
+        raise BAD_REQUEST
+
+
+# modify album
+@router.patch("", description="Modify album")
+async def modify_album(
+    supabase: Annotated[Client, Depends(get_supabase)],
+    user_id: Annotated[str, Security(get_id)],
+    album: Album,
+    songs: List[int],
+):
+    try:
+        if album.user_id != user_id:
+            raise HTTPException(
+                status_code=status.HTTP_400_BAD_REQUEST,
+                detail="You are not the owner of this album",
+            )
+        supabase.table("albums").update(
+            jsonable_encoder(album, exclude={"id", "user_id", "created_at"})
+        ).eq("id", album.id).execute()
+        supabase.table("songs").update({"album_id": None}).eq(
+            "album_id", album.id
+        ).execute()
+        supabase.table("songs").update({"album_id": album.id}).in_(
+            "id", songs
+        ).execute()
+        return {"detail": "Album modified"}
+    except:
+        raise BAD_REQUEST
+
+
+# get album and its songs
+@router.get("", description="Get album and its songs")
+async def get_album(
+    supabase: Annotated[Client, Depends(get_supabase)],
+    user_id: Annotated[str, Security(get_id)],
+    album_id: int,
+):
+    try:
+        res = (
+            supabase.table("albums")
+            .select("*, songs(*)")
+            .eq("id", album_id)
+            .execute()
+            .dict()["data"][0]
+        )
+        return res
+    except:
+        raise BAD_REQUEST

app/routers/authentication.py

@@ -0,0 +1,114 @@
+from fastapi import APIRouter, Depends, HTTPException, status, Security, Body, Query
+from fastapi.security import OAuth2PasswordRequestForm
+from models.user import UserSignup
+from db.supabase_service import get_supabase
+from typing import Annotated
+from supabase import Client
+from utils.auth import get_id
+from fastapi.encoders import jsonable_encoder
+from models.enums import Role
+from gotrue.errors import AuthApiError
+from crud.user import update_user
+from utils.exceptions import BAD_REQUEST, CONFLICT, NOT_FOUND
+
+router = APIRouter(tags=["Authentication"])
+
+
+@router.post("/login", description="Login user using email and password")
+async def login(
+    form_data: Annotated[OAuth2PasswordRequestForm, Depends()],
+    supabase: Annotated[Client, Depends(get_supabase)],
+):
+    email = form_data.username
+    password = form_data.password
+    try:
+        user = supabase.auth.sign_in_with_password(
+            {"email": email, "password": password}
+        )
+        return {
+            "access_token": user.session.access_token,
+            "token_type": "bearer",
+        }
+    except:
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail="Incorrect email or password",
+        )
+
+
+@router.post(
+    "/signup", description="Sign up new user", status_code=status.HTTP_201_CREATED
+)
+async def signup(
+    user: Annotated[UserSignup, Body()],
+    supabase: Annotated[Client, Depends(get_supabase)],
+):
+    email = user.email
+    password = user.password
+    if len(password) < 6:
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail="Password should be at least 6 characters",
+        )
+    try:
+        user.role = Role.user
+        res = supabase.auth.sign_up(
+            {
+                "email": email,
+                "password": password,
+                "options": {
+                    "data": jsonable_encoder(user, exclude=("password", "email"))
+                },
+            }
+        )
+        return {"detail": "User created"}
+    except AuthApiError:
+        raise CONFLICT
+    except:
+        raise BAD_REQUEST
+
+
+@router.post("/reset_password", description="Send reset password email")
+async def reset_password(
+    supabase: Annotated[Client, Depends(get_supabase)],
+    email: str,
+):
+    try:
+        if (
+            len(supabase.rpc("get_user_id_by_email", {"email": email}).execute().data)
+            > 0
+        ):
+            supabase.auth.reset_password_email(email)
+            return {"detail": "Reset password email sent"}
+        else:
+            raise NOT_FOUND
+    except:
+        raise NOT_FOUND
+
+
+@router.post("/reset_password_confirm", description="Reset password")
+async def reset_password_confirm(
+    supabase: Annotated[Client, Depends(get_supabase)],
+    email: Annotated[str, Query(description="Email", title="Email")],
+    new_password: Annotated[
+        str, Query(min_length=6, description="New password", title="New password")
+    ],
+    token: Annotated[
+        str, Query(description="Reset password token", title="Reset password token")
+    ],
+):
+    try:
+        supabase.auth.verify_otp(
+            {
+                "email": email,
+                "token": token,
+                "type": "email",
+            }
+        )
+        await update_user(supabase, password=new_password)
+        return {"detail": "Your password has been reset"}
+    except:
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail="Incorrect token",
+        )

app/routers/concert.py

@@ -0,0 +1,69 @@
+from fastapi import APIRouter, Depends, Security
+from typing import Annotated
+from fastapi.encoders import jsonable_encoder
+from supabase import Client
+from db.supabase_service import get_supabase
+from typing import Annotated
+from supabase import Client
+from utils.auth import get_id
+from utils.exceptions import BAD_REQUEST
+from models.concert import ConcertInfo
+
+router = APIRouter(tags=["Concert"], prefix="/concert")
+
+
+# create concert
+@router.post("", description="Create concert")
+async def create_concert(
+    supabase: Annotated[Client, Depends(get_supabase)],
+    user_id: Annotated[str, Security(get_id)],
+    concert_info: ConcertInfo,
+):
+    try:
+        supabase.table("concerts").insert(jsonable_encoder(concert_info)).execute()
+        return {"detail": "Concert created"}
+    except:
+        raise BAD_REQUEST
+
+
+# get list of all concert
+@router.get("/all", description="Get list of all concert")
+async def get_all_concert(
+    supabase: Annotated[Client, Depends(get_supabase)],
+    user_id: Annotated[str, Security(get_id)],
+):
+    try:
+        res = supabase.table("concerts").select("*").execute().dict()["data"]
+        return res
+    except:
+        raise BAD_REQUEST
+
+
+# modify concert
+@router.put("", description="Modify concert")
+async def modify_concert(
+    supabase: Annotated[Client, Depends(get_supabase)],
+    user_id: Annotated[str, Security(get_id)],
+    concert_info: ConcertInfo,
+):
+    try:
+        supabase.table("concerts").update(jsonable_encoder(concert_info)).eq(
+            "id", concert_info.id
+        ).execute()
+        return {"detail": "Concert modified"}
+    except:
+        raise BAD_REQUEST
+
+
+# delete concert
+@router.delete("", description="Delete concert")
+async def delete_concert(
+    supabase: Annotated[Client, Depends(get_supabase)],
+    user_id: Annotated[str, Security(get_id)],
+    concert_id: int,
+):
+    try:
+        supabase.table("concerts").delete().eq("id", concert_id).execute()
+        return {"detail": "Concert deleted"}
+    except:
+        raise BAD_REQUEST

app/routers/group_admin.py

@@ -0,0 +1,107 @@
+from fastapi import APIRouter, Depends, HTTPException, status, Security
+from db.supabase_service import get_supabase
+from typing import Annotated
+from supabase import Client
+from utils.auth import get_role, get_id, oauth2_scheme
+from models.enums import Role
+from models.group import GroupInfo
+from utils.exceptions import UNAUTHORIZED
+from typing import List
+
+router = APIRouter(tags=["Group"], prefix="/group")
+
+
+@router.post("", description="Create group")
+async def create_group(
+    supabase: Annotated[Client, Depends(get_supabase)],
+    role: Annotated[str, Security(get_role)],
+    id: Annotated[str, Security(get_id)],
+    group_info: GroupInfo,
+):
+    if role != Role.group_admin:
+        raise UNAUTHORIZED
+    supabase.table("groups").insert(
+        {
+            "name": group_info.name,
+            "description": group_info.description,
+            "admin_id": id,
+        }
+    ).execute()
+    return {"detail": "Group created"}
+
+
+@router.post("/user", description="Modify group")
+async def add_user(
+    token: Annotated[str, Depends(oauth2_scheme)],
+    supabase: Annotated[Client, Depends(get_supabase)],
+    role: Annotated[str, Security(get_role)],
+    id: Annotated[str, Security(get_id)],
+    email: str,
+    group_id: int,
+):
+    if role != Role.group_admin:
+        raise UNAUTHORIZED
+    user_id = supabase.rpc("get_user_id_by_email", {"email": email}).execute().data
+    if len(user_id) == 0:
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail="User does not exist",
+        )
+    user_id = user_id[0]["id"]
+    members_id: List = (
+        supabase.table("groups")
+        .select("members_id")
+        .match({"admin_id": id, "id": group_id})
+        .execute()
+        .data[0]["members_id"]
+    )
+    if not members_id:
+        members_id = []
+    if user_id in members_id:
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail="User already in group",
+        )
+    members_id.append(user_id)
+    supabase.table("groups").update({"members_id": members_id}).match(
+        {"admin_id": id, "id": group_id}
+    ).execute()
+    return {"detail": "User added to group"}
+
+
+@router.delete("/user", description="Delete user from group")
+async def delete_user(
+    supabase: Annotated[Client, Depends(get_supabase)],
+    role: Annotated[str, Security(get_role)],
+    id: Annotated[str, Security(get_id)],
+    group_id: int,
+    email: str,
+):
+    if role != Role.group_admin:
+        raise UNAUTHORIZED
+    user_id = supabase.rpc("get_user_id_by_email", {"email": email}).execute().data
+    if len(user_id) == 0:
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail="User does not exist",
+        )
+    user_id = user_id[0]["id"]
+    members_id: List = (
+        supabase.table("groups")
+        .select("members_id")
+        .match({"admin_id": id, "id": group_id})
+        .execute()
+        .data[0]["members_id"]
+    )
+    if not members_id:
+        members_id = []
+    if user_id not in members_id:
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail="User not in group",
+        )
+    members_id.remove(user_id)
+    supabase.table("groups").update({"members_id": members_id}).match(
+        {"admin_id": id, "id": group_id}
+    ).execute()
+    return {"detail": "User deleted from group"}

app/routers/playlist.py

@@ -0,0 +1,104 @@
+from fastapi import (
+    APIRouter,
+    Depends,
+    HTTPException,
+    status,
+    Security,
+)
+from models.music import Playlist, BasePlaylist
+from db.supabase_service import get_supabase
+from typing import Annotated, List
+from supabase import Client
+from utils.auth import get_id
+from fastapi.encoders import jsonable_encoder
+from utils.exceptions import BAD_REQUEST
+
+router = APIRouter(tags=["Playlist"], prefix="/playlist")
+
+
+@router.post("", description="Create playlist")
+async def create_playlist(
+    supabase: Annotated[Client, Depends(get_supabase)],
+    user_id: Annotated[str, Security(get_id)],
+    playlist: BasePlaylist,
+):
+    try:
+        supabase.table("playlist").insert(
+            {"name": playlist.name, "song_id" : [], "user_id": user_id}
+        ).execute()
+        return {"detail": "Playlist created"}
+    except:
+        raise BAD_REQUEST
+    
+
+@router.delete("", description="Delete playlist")
+async def delete_playlist(
+    supabase: Annotated[Client, Depends(get_supabase)],
+    user_id: Annotated[str, Security(get_id)],
+    playlist_id: int,
+):
+    try:
+        supabase.table("playlist").delete().eq("id", playlist_id).execute()
+        return {"detail": "Album deleted"}
+    except:
+        raise BAD_REQUEST
+
+# get list of all your playlist      
+@router.get("/all", description="Get list of your playlist", response_model=List[Playlist])
+async def get_all_album(
+    supabase: Annotated[Client, Depends(get_supabase)],
+    user_id: Annotated[str, Security(get_id)],
+) -> List[Playlist]:
+    try:
+        
+        res = supabase.table("playlist").select("*").match({"user_id" : user_id}).execute().data
+        return res
+    except:
+        raise BAD_REQUEST
+
+# modify playlist
+@router.patch("", description="Modify playlist")
+async def modify_playlist(
+    supabase: Annotated[Client, Depends(get_supabase)],
+    user_id: Annotated[str, Security(get_id)],
+    playlist: Playlist,
+    songs: List[int],
+):
+    
+    if playlist.user_id != user_id:
+            raise HTTPException(
+                status_code=status.HTTP_400_BAD_REQUEST,
+                detail="You are not the owner of this playlist",
+            )
+    try:
+        supabase.table("playlist").update(
+            jsonable_encoder(playlist, exclude={"id", "user_id", "created_at"})
+        ).eq("id", playlist.id).execute()
+
+        supabase.table("playlist").update({ "song_id": songs }).eq("id", playlist.id).execute()
+
+        return {"detail": "Playlist modified"}
+    except:
+        raise BAD_REQUEST
+    
+
+
+
+# get playlist and its songs
+@router.get("/{playlist_id}", description="Get playlist and its songs")
+async def get_playlist(
+    supabase: Annotated[Client, Depends(get_supabase)],
+    user_id: Annotated[str, Security(get_id)],
+    playlist_id: int,
+):
+    try:
+        res = (
+            supabase.table("playlist")
+            .select("*")
+            .eq("id", playlist_id)
+            .execute()
+            .dict()["data"]
+        )
+        return res
+    except:
+        raise BAD_REQUEST

app/routers/song.py

@@ -0,0 +1,129 @@
+from fastapi import (
+    APIRouter,
+    Depends,
+    HTTPException,
+    UploadFile,
+    status,
+    Security,
+)
+from models.music import Song
+from db.supabase_service import get_supabase
+from typing import Annotated, List, Tuple
+from supabase import Client
+from utils.auth import get_id
+from fastapi.encoders import jsonable_encoder
+from utils.exceptions import BAD_REQUEST
+
+router = APIRouter(tags=["Song"], prefix="/song")
+
+
+# upload mp3 file to supabase storage
+@router.post("", description="Upload music file")
+async def upload_music(
+    supabase: Annotated[Client, Depends(get_supabase)],
+    id: Annotated[str, Security(get_id)],
+    music: UploadFile,
+    name: str | None = None,
+):
+    if music.content_type != "audio/mpeg":
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail="Only mp3 file is accepted",
+        )
+    try:
+        if not name:
+            name = music.filename
+        f = await music.read()
+        supabase.storage.from_("songs").upload(
+            path=music.filename, file=f, file_options={"content-type": "audio/mpeg"}
+        )
+        supabase.table("songs").insert(
+            {"name": name, "name_in_storage": music.filename, "user_id": id}
+        ).execute()
+        return {"detail": "Music uploaded"}
+    except:
+        raise BAD_REQUEST
+
+
+# get url of music file from supabase storage
+@router.get("/link", description="Get url of music file")
+async def get_link_to_music(
+    supabase: Annotated[Client, Depends(get_supabase)],
+    id: Annotated[str, Security(get_id)],
+    name_in_storage: str,
+):
+    try:
+        res = supabase.storage.from_("songs").create_signed_url(
+            path=name_in_storage, expires_in=3600
+        )
+        return res
+    except:
+        raise BAD_REQUEST
+
+
+# get list of all song from search
+@router.get(
+    "/search", description="Get list of all music", response_model=List[Song]
+)
+async def get_music(
+    supabase: Annotated[Client, Depends(get_supabase)],
+    id: Annotated[str, Security(get_id)],
+    query: str | None = None,
+) -> List[Song]:
+    try:
+        if query:
+            res = (
+                supabase.table("songs")
+                .select("*")
+                .ilike("name", f"%{query}%")
+                .execute()
+                .dict()["data"]
+            )
+        else:
+            res = supabase.table("songs").select("*").execute().dict()["data"]
+        return res
+    except:
+        raise BAD_REQUEST
+
+
+# delete song
+@router.delete("", description="Delete music")
+async def delete_music(
+    supabase: Annotated[Client, Depends(get_supabase)],
+    user_id: Annotated[str, Security(get_id)],
+    song: Song,
+):
+    try:
+        if song.user_id != user_id:
+            raise HTTPException(
+                status_code=status.HTTP_400_BAD_REQUEST,
+                detail="You are not the owner of this music",
+            )
+        supabase.storage.from_("songs").remove([song.name_in_storage])
+        supabase.table("songs").delete().eq("id", song.id).execute()
+        return {"detail": "Music deleted"}
+    except:
+        raise
+
+
+# change music info
+@router.patch("/info", description="Change music info")
+async def change_music_info(
+    supabase: Annotated[Client, Depends(get_supabase)],
+    user_id: Annotated[str, Security(get_id)],
+    song: Song,
+):
+    try:
+        if song.user_id != user_id:
+            raise HTTPException(
+                status_code=status.HTTP_400_BAD_REQUEST,
+                detail="You are not the owner of this music",
+            )
+        supabase.table("songs").update(
+            jsonable_encoder(
+                song, exclude={"id", "user_id", "created_at", "name_in_storage"}
+            )
+        ).eq("id", song.id).execute()
+        return {"detail": "Music info updated"}
+    except:
+        raise BAD_REQUEST

app/routers/users.py

@@ -0,0 +1,134 @@
+from fastapi import APIRouter, Depends, HTTPException, status, Security, Body, Query
+from db.supabase_service import get_supabase
+from typing import Annotated
+from supabase import Client
+from utils.auth import get_role, get_id, oauth2_scheme
+from models.enums import Role
+from models.user import UserInfo
+from models.group import GroupInfo
+from fastapi.encoders import jsonable_encoder
+from crud.user import get_user, update_user
+from utils.exceptions import BAD_REQUEST
+
+router = APIRouter(tags=["User"], prefix="/user")
+
+
+@router.get("/info", description="Get information of logged in user")
+async def get_user_info(
+    token: Annotated[str, Depends(oauth2_scheme)],
+    supabase: Annotated[Client, Depends(get_supabase)],
+):
+    return await get_user(supabase, token)
+
+
+@router.post(
+    "/signup_group_admin",
+    description="Sign up to be a group admin",
+    status_code=status.HTTP_201_CREATED,
+)
+async def signup_as_group_admin(
+    supabase: Annotated[Client, Depends(get_supabase)],
+    role: Annotated[str, Security(get_role)],
+    id: Annotated[str, Security(get_id)],
+    group_info: GroupInfo,
+):
+    if role != Role.user:
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail="You are already a group admin or an admin",
+        )
+    try:
+        supabase.auth.update_user({"data": {"role": Role.group_admin}})
+        supabase.table("groups").insert(
+            {
+                "name": group_info.name,
+                "description": group_info.description,
+                "admin_id": id,
+            }
+        ).execute()
+        return {"detail": "You are now a group admin"}
+    except:
+        raise BAD_REQUEST
+
+
+@router.put("/password", description="Change password of logged in user")
+async def change_password(
+    supabase: Annotated[Client, Depends(get_supabase)],
+    id: Annotated[str, Security(get_id)],
+    new_password: Annotated[
+        str, Query(min_length=6, description="New password", title="New password")
+    ],
+):
+    try:
+        await update_user(supabase, password=new_password)
+        return {"detail": "Password updated"}
+    except:
+        raise
+
+
+@router.patch("/info", description="Update user info")
+async def update_user_info(
+    supabase: Annotated[Client, Depends(get_supabase)],
+    id: Annotated[str, Security(get_id)],
+    new_user_info: Annotated[UserInfo, Body()],
+):
+    try:
+        if new_user_info.role:
+            del new_user_info.role
+        await update_user(supabase, data=jsonable_encoder(new_user_info))
+        return {"detail": "User info updated"}
+    except:
+        raise BAD_REQUEST
+
+
+#  buy ticket
+@router.post("/buy_ticket", description="Buy ticket")
+async def buy_ticket(
+    supabase: Annotated[Client, Depends(get_supabase)],
+    user_id: Annotated[str, Security(get_id)],
+    concert_id: int,
+):
+    try:
+        number_of_tickets = (
+            supabase.table("concerts")
+            .select("number_of_tickets")
+            .eq("id", concert_id)
+            .execute()
+            .dict()["data"][0]["number_of_tickets"]
+        )
+        if number_of_tickets == 0:
+            raise HTTPException(
+                status_code=status.HTTP_400_BAD_REQUEST,
+                detail="There are no tickets left",
+            )
+        return {"detail": "Please navigate to the payment page"}
+    except:
+        raise BAD_REQUEST
+
+
+# Confirm payment
+@router.post("/confirm_payment", description="Confirm payment")
+async def confirm_payment(
+    supabase: Annotated[Client, Depends(get_supabase)],
+    user_id: Annotated[str, Security(get_id)],
+    concert_id: int,
+):
+    try:
+        number_of_tickets = (
+            supabase.table("concerts")
+            .select("number_of_tickets")
+            .eq("id", concert_id)
+            .execute()
+            .dict()["data"][0]["number_of_tickets"]
+        )
+        if number_of_tickets == 0:
+            raise HTTPException(
+                status_code=status.HTTP_400_BAD_REQUEST,
+                detail="There are no tickets left",
+            )
+        supabase.table("concerts").update(
+            {"number_of_tickets": number_of_tickets - 1}
+        ).eq("id", concert_id).execute()
+        return {"detail": "Payment successful"}
+    except:
+        raise BAD_REQUEST

app/tests/__init__.py

@@ -0,0 +1 @@
+# This file is intentionally left blank.

app/tests/test_auth.py

@@ -0,0 +1,2 @@
+import pytest
+from fastapi.testclient import TestClient