247 lines
4.1 KiB
Python
247 lines
4.1 KiB
Python
from fastapi import FastAPI, Depends
|
|
from pydantic import BaseModel
|
|
|
|
from database import fetch_all, fetch_one
|
|
from events import create_event
|
|
from auth import create_user, verify_user, create_token, get_current_user
|
|
from rooms import create_room, search_room, accept_room
|
|
|
|
|
|
class Login(BaseModel):
|
|
username: str
|
|
password: str
|
|
|
|
|
|
class UserCreate(BaseModel):
|
|
username: str
|
|
password: str
|
|
|
|
|
|
class EventCreate(BaseModel):
|
|
room_id: int
|
|
type: str
|
|
title: str
|
|
description: str
|
|
start_time: str
|
|
end_time: str
|
|
visibility: str
|
|
status: str
|
|
min_people: int
|
|
|
|
|
|
class RoomCreate(BaseModel):
|
|
room_name: str
|
|
invite_code: str
|
|
|
|
|
|
class RoomSearch(BaseModel):
|
|
invite_code: str
|
|
|
|
|
|
class RoomJoin(BaseModel):
|
|
invite_code: str
|
|
|
|
|
|
app = FastAPI(
|
|
title="Calendar API",
|
|
description="Room based agenda system",
|
|
version="1.0"
|
|
)
|
|
|
|
|
|
@app.get("/")
|
|
def home():
|
|
return {
|
|
"status": "online",
|
|
"service": "Calendar API"
|
|
}
|
|
|
|
|
|
@app.get("/database")
|
|
def database_status():
|
|
tables = fetch_all(
|
|
"SELECT name FROM sqlite_master WHERE type='table';"
|
|
)
|
|
|
|
return {
|
|
"tables": [table["name"] for table in tables]
|
|
}
|
|
|
|
|
|
# users
|
|
|
|
|
|
@app.post("/create-user")
|
|
def api_create_user(user: UserCreate):
|
|
|
|
user_id = create_user(
|
|
user.username,
|
|
user.password
|
|
)
|
|
|
|
return {
|
|
"status": "created",
|
|
"user_id": user_id
|
|
}
|
|
|
|
|
|
@app.post("/login")
|
|
def api_login(login: Login):
|
|
|
|
user_id = verify_user(
|
|
login.username,
|
|
login.password
|
|
)
|
|
|
|
if user_id is None:
|
|
return {
|
|
"status": "failed"
|
|
}
|
|
|
|
token = create_token(user_id)
|
|
|
|
return {
|
|
"status": "success",
|
|
"access_token": token,
|
|
"token_type": "bearer"
|
|
}
|
|
|
|
|
|
@app.get("/me")
|
|
def get_me(
|
|
user_id: int = Depends(get_current_user)
|
|
):
|
|
return {
|
|
"user_id": user_id
|
|
}
|
|
|
|
|
|
# rooms
|
|
|
|
|
|
@app.post("/create-room")
|
|
def api_create_room(
|
|
room: RoomCreate,
|
|
user_id: int = Depends(get_current_user)
|
|
):
|
|
|
|
room_id = create_room(
|
|
room.room_name,
|
|
user_id,
|
|
room.invite_code
|
|
)
|
|
|
|
return {
|
|
"status": "created",
|
|
"room_id": room_id
|
|
}
|
|
|
|
|
|
@app.post("/search-room")
|
|
def api_search_room(
|
|
room: RoomSearch
|
|
):
|
|
|
|
room_name = search_room(
|
|
room.invite_code
|
|
)
|
|
|
|
if room_name is None:
|
|
return {
|
|
"status": "failed",
|
|
"reason": "room_not_found"
|
|
}
|
|
|
|
return {
|
|
"status": "found",
|
|
"room_name": room_name
|
|
}
|
|
|
|
|
|
@app.post("/join-room")
|
|
def api_join_room(
|
|
room: RoomJoin,
|
|
user_id: int = Depends(get_current_user)
|
|
):
|
|
|
|
result = accept_room(
|
|
user_id,
|
|
room.invite_code
|
|
)
|
|
|
|
if result is None:
|
|
return {
|
|
"status": "failed",
|
|
"reason": "room_not_found"
|
|
}
|
|
|
|
return {
|
|
"status": "joined",
|
|
**result
|
|
}
|
|
|
|
# events
|
|
|
|
@app.post("/events")
|
|
def api_create_event(
|
|
event: EventCreate,
|
|
user_id: int = Depends(get_current_user)
|
|
):
|
|
|
|
# check of gebruiker lid is
|
|
member = fetch_one(
|
|
"""
|
|
SELECT *
|
|
FROM room_members
|
|
WHERE room_id = ?
|
|
AND user_id = ?
|
|
""",
|
|
(
|
|
event.room_id,
|
|
user_id
|
|
)
|
|
)
|
|
|
|
if member is None:
|
|
return {
|
|
"status": "failed",
|
|
"reason": "not_room_member"
|
|
}
|
|
|
|
|
|
event_id = create_event(
|
|
room_id=event.room_id,
|
|
creator_id=user_id,
|
|
type=event.type,
|
|
title=event.title,
|
|
description=event.description,
|
|
start_time=event.start_time,
|
|
end_time=event.end_time,
|
|
visibility=event.visibility,
|
|
status=event.status,
|
|
min_people=event.min_people
|
|
)
|
|
|
|
return {
|
|
"status": "created",
|
|
"event_id": event_id
|
|
}
|
|
|
|
|
|
@app.get("/events")
|
|
def get_events(
|
|
user_id: int = Depends(get_current_user)
|
|
):
|
|
|
|
events = fetch_all(
|
|
"""
|
|
SELECT *
|
|
FROM events
|
|
WHERE creator_id = ?
|
|
""",
|
|
(user_id,)
|
|
)
|
|
|
|
return {
|
|
"events": [dict(event) for event in events]
|
|
} |