724 lines
14 KiB
Python
724 lines
14 KiB
Python
from fastapi import FastAPI, Depends
|
|
from pydantic import BaseModel
|
|
|
|
from database import fetch_all, fetch_one
|
|
from auth import create_user, verify_user, create_token, get_current_user
|
|
from rooms import create_room, search_room, accept_room, get_rooms, delete_room, get_room
|
|
from permissions import has_role
|
|
from events import create_event, get_user_events, check_event_threshold
|
|
|
|
from event_signups import respond_to_event, count_confirmations
|
|
|
|
from fastapi import HTTPException
|
|
import sqlite3
|
|
from typing import Literal
|
|
from fastapi.middleware.cors import CORSMiddleware
|
|
|
|
from events import delete_event
|
|
|
|
class Login(BaseModel):
|
|
username: str
|
|
password: str
|
|
|
|
class GetRooms(BaseModel):
|
|
user_id: 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
|
|
|
|
class EventUpdate(BaseModel):
|
|
title: str | None = None
|
|
description: str | None = None
|
|
start_time: str | None = None
|
|
end_time: str | None = None
|
|
min_people: int | None = None
|
|
|
|
|
|
app = FastAPI(
|
|
title="Calendar API",
|
|
description="Room based agenda system",
|
|
version="1.0"
|
|
)
|
|
|
|
app.add_middleware(
|
|
CORSMiddleware,
|
|
allow_origins=["*"], # later vervangen door je domein
|
|
allow_credentials=True,
|
|
allow_methods=["*"],
|
|
allow_headers=["*"],
|
|
)
|
|
|
|
|
|
@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):
|
|
|
|
try:
|
|
user_id = create_user(
|
|
user.username,
|
|
user.password
|
|
)
|
|
|
|
except sqlite3.IntegrityError:
|
|
raise HTTPException(
|
|
status_code=409,
|
|
detail="username already exists"
|
|
)
|
|
|
|
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)
|
|
):
|
|
|
|
user = fetch_one(
|
|
"""
|
|
SELECT id, username
|
|
FROM users
|
|
WHERE id = ?
|
|
""",
|
|
(user_id,)
|
|
)
|
|
|
|
return dict(user)
|
|
|
|
# rooms
|
|
|
|
@app.delete("/room/{room_id}")
|
|
def delete_room(
|
|
room_id: int,
|
|
user_id: int = Depends(get_current_user)
|
|
):
|
|
|
|
if not has_role(
|
|
room_id,
|
|
user_id,
|
|
["owner", "admin"]
|
|
):
|
|
return {
|
|
"status": "denied",
|
|
"reason": "no_permission"
|
|
}
|
|
return {
|
|
"status": "not deleted",
|
|
"reason": "you are a poopie",
|
|
"actual Reason": "i have not built it yet"
|
|
}
|
|
|
|
|
|
# @app.post("/create-room")
|
|
# def api_create_room(
|
|
# room: RoomCreate,
|
|
# user_id: int = Depends(get_current_user)
|
|
# ):
|
|
|
|
# room_id = create_room(
|
|
# user_id,
|
|
# room.room_name,
|
|
# room.invite_code
|
|
# )
|
|
|
|
# return {
|
|
# "status": "created",
|
|
# **room_id
|
|
# }
|
|
|
|
@app.post("/create-room")
|
|
def api_create_room(
|
|
room: RoomCreate,
|
|
user_id: int = Depends(get_current_user)
|
|
):
|
|
|
|
try:
|
|
room_id = create_room(
|
|
user_id,
|
|
room.room_name,
|
|
room.invite_code
|
|
)
|
|
|
|
except sqlite3.IntegrityError:
|
|
raise HTTPException(
|
|
status_code=409,
|
|
detail="invite_code or roomname already exists"
|
|
)
|
|
|
|
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)
|
|
):
|
|
|
|
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 = get_user_events(user_id)
|
|
|
|
return {
|
|
"events": [dict(event) for event in events]
|
|
}
|
|
|
|
|
|
@app.post("/events/{event_id}/respond")
|
|
def respond_event(
|
|
event_id: int,
|
|
status: Literal["going", "maybe", "declined"],
|
|
user_id: int = Depends(get_current_user)
|
|
):
|
|
|
|
event = fetch_one(
|
|
"""
|
|
SELECT room_id
|
|
FROM events
|
|
WHERE id = ?
|
|
""",
|
|
(event_id,)
|
|
)
|
|
|
|
if event is None:
|
|
raise HTTPException(
|
|
status_code=404,
|
|
detail="event not found"
|
|
)
|
|
|
|
member = fetch_one(
|
|
"""
|
|
SELECT *
|
|
FROM room_members
|
|
WHERE room_id = ?
|
|
AND user_id = ?
|
|
""",
|
|
(
|
|
event["room_id"],
|
|
user_id
|
|
)
|
|
)
|
|
|
|
if member is None:
|
|
raise HTTPException(
|
|
status_code=403,
|
|
detail="not a room member"
|
|
)
|
|
|
|
|
|
respond_to_event(
|
|
event_id,
|
|
user_id,
|
|
status
|
|
)
|
|
|
|
check_event_threshold(event_id)
|
|
|
|
return {
|
|
"status": "response_saved",
|
|
"going": count_confirmations(event_id)
|
|
}
|
|
|
|
@app.get("/rooms")
|
|
def api_get_rooms(
|
|
user_id: int = Depends(get_current_user)
|
|
):
|
|
rooms = get_rooms(user_id)
|
|
|
|
return {
|
|
"rooms": [dict(room) for room in rooms]
|
|
}
|
|
|
|
@app.get("/rooms/{room_id}")
|
|
def api_get_room(
|
|
room_id: int,
|
|
user_id: int = Depends(get_current_user)
|
|
):
|
|
|
|
room = get_room(
|
|
room_id,
|
|
user_id
|
|
)
|
|
|
|
if room is None:
|
|
raise HTTPException(
|
|
status_code=404,
|
|
detail="room not found or no permission"
|
|
)
|
|
|
|
return room
|
|
|
|
@app.get("/rooms/{room_id}/members")
|
|
def api_room_members(
|
|
room_id: int,
|
|
user_id: int = Depends(get_current_user)
|
|
):
|
|
|
|
room = get_room(room_id, user_id)
|
|
|
|
if room is None:
|
|
raise HTTPException(
|
|
status_code=404,
|
|
detail="room not found or no permission"
|
|
)
|
|
|
|
members = get_room_members(room_id)
|
|
|
|
return {
|
|
"members": [dict(member) for member in members]
|
|
}
|
|
|
|
@app.get("/events/{event_id}")
|
|
def api_get_event(
|
|
event_id: int,
|
|
user_id: int = Depends(get_current_user)
|
|
):
|
|
|
|
event = fetch_one(
|
|
"""
|
|
SELECT *
|
|
FROM events
|
|
WHERE id = ?
|
|
""",
|
|
(event_id,)
|
|
)
|
|
|
|
if event is None:
|
|
raise HTTPException(
|
|
status_code=404,
|
|
detail="event not found"
|
|
)
|
|
|
|
member = fetch_one(
|
|
"""
|
|
SELECT *
|
|
FROM room_members
|
|
WHERE room_id = ?
|
|
AND user_id = ?
|
|
""",
|
|
(
|
|
event["room_id"],
|
|
user_id
|
|
)
|
|
)
|
|
|
|
if member is None:
|
|
raise HTTPException(
|
|
status_code=403,
|
|
detail="no access"
|
|
)
|
|
|
|
return dict(event)
|
|
|
|
@app.get("/events/{event_id}/responses")
|
|
def api_event_responses(
|
|
event_id: int,
|
|
user_id: int = Depends(get_current_user)
|
|
):
|
|
|
|
event = fetch_one(
|
|
"""
|
|
SELECT room_id
|
|
FROM events
|
|
WHERE id = ?
|
|
""",
|
|
(event_id,)
|
|
)
|
|
|
|
if event is None:
|
|
raise HTTPException(
|
|
status_code=404,
|
|
detail="event not found"
|
|
)
|
|
|
|
member = fetch_one(
|
|
"""
|
|
SELECT *
|
|
FROM room_members
|
|
WHERE room_id = ?
|
|
AND user_id = ?
|
|
""",
|
|
(
|
|
event["room_id"],
|
|
user_id
|
|
)
|
|
)
|
|
|
|
if member is None:
|
|
raise HTTPException(
|
|
status_code=403,
|
|
detail="no access"
|
|
)
|
|
|
|
responses = get_event_responses(event_id)
|
|
|
|
return {
|
|
"responses": [dict(r) for r in responses]
|
|
}
|
|
|
|
@app.get("/users/{user_id}")
|
|
def api_get_user(
|
|
user_id: int,
|
|
current_user: int = Depends(get_current_user)
|
|
):
|
|
|
|
user = fetch_one(
|
|
"""
|
|
SELECT id, username
|
|
FROM users
|
|
WHERE id = ?
|
|
""",
|
|
(user_id,)
|
|
)
|
|
|
|
if user is None:
|
|
raise HTTPException(
|
|
status_code=404,
|
|
detail="user not found"
|
|
)
|
|
|
|
return dict(user)
|
|
|
|
@app.delete("/events/{event_id}")
|
|
def api_delete_event(
|
|
event_id: int,
|
|
user_id: int = Depends(get_current_user)
|
|
):
|
|
|
|
event = fetch_one(
|
|
"""
|
|
SELECT creator_id
|
|
FROM events
|
|
WHERE id = ?
|
|
""",
|
|
(event_id,)
|
|
)
|
|
|
|
if event is None:
|
|
raise HTTPException(
|
|
status_code=404,
|
|
detail="event not found"
|
|
)
|
|
|
|
if event["creator_id"] != user_id:
|
|
raise HTTPException(
|
|
status_code=403,
|
|
detail="not event owner"
|
|
)
|
|
|
|
delete_event(event_id)
|
|
|
|
return {
|
|
"status": "deleted"
|
|
}
|
|
|
|
@app.patch("/events/{event_id}")
|
|
def api_update_event(
|
|
event_id: int,
|
|
event: EventUpdate,
|
|
user_id: int = Depends(get_current_user)
|
|
):
|
|
|
|
old = fetch_one(
|
|
"""
|
|
SELECT creator_id
|
|
FROM events
|
|
WHERE id = ?
|
|
""",
|
|
(event_id,)
|
|
)
|
|
|
|
if old is None:
|
|
raise HTTPException(
|
|
status_code=404,
|
|
detail="event not found"
|
|
)
|
|
|
|
if old["creator_id"] != user_id:
|
|
raise HTTPException(
|
|
status_code=403,
|
|
detail="not owner"
|
|
)
|
|
|
|
|
|
update_event(
|
|
event_id,
|
|
event.title,
|
|
event.description,
|
|
event.start_time,
|
|
event.end_time,
|
|
event.min_people
|
|
)
|
|
|
|
|
|
return {
|
|
"status": "updated"
|
|
}
|
|
|
|
@app.delete("/rooms/{room_id}/members/me")
|
|
def api_leave_room(
|
|
room_id: int,
|
|
user_id: int = Depends(get_current_user)
|
|
):
|
|
|
|
member = fetch_one(
|
|
"""
|
|
SELECT *
|
|
FROM room_members
|
|
WHERE room_id = ?
|
|
AND user_id = ?
|
|
""",
|
|
(
|
|
room_id,
|
|
user_id
|
|
)
|
|
)
|
|
|
|
if member is None:
|
|
raise HTTPException(
|
|
status_code=404,
|
|
detail="not member"
|
|
)
|
|
|
|
|
|
leave_room(
|
|
room_id,
|
|
user_id
|
|
)
|
|
|
|
|
|
return {
|
|
"status": "left"
|
|
}
|
|
|
|
@app.get("/rooms/{room_id}/events")
|
|
def api_room_events(
|
|
room_id: int,
|
|
user_id: int = Depends(get_current_user)
|
|
):
|
|
|
|
member = fetch_one(
|
|
"""
|
|
SELECT *
|
|
FROM room_members
|
|
WHERE room_id = ?
|
|
AND user_id = ?
|
|
""",
|
|
(
|
|
room_id,
|
|
user_id
|
|
)
|
|
)
|
|
|
|
if member is None:
|
|
raise HTTPException(
|
|
status_code=403,
|
|
detail="no access"
|
|
)
|
|
|
|
|
|
events = get_room_events(room_id)
|
|
|
|
|
|
return {
|
|
"events": [
|
|
dict(event)
|
|
for event in events
|
|
]
|
|
}
|
|
|
|
from fastapi.responses import PlainTextResponse
|
|
import requests
|
|
|
|
|
|
@app.get("/ascii-art", response_class=PlainTextResponse)
|
|
def ascii_art():
|
|
response = requests.get(
|
|
"http://localhost:8000/me"
|
|
)
|
|
|
|
data = response.json()
|
|
me_endpoint = data
|
|
status = "online"
|
|
|
|
return f"""
|
|
██████╗ █████╗ ██╗ ███████╗███╗ ██╗██████╗ █████╗ ██████╗
|
|
██╔════╝██╔══██╗██║ ██╔════╝████╗ ██║██╔══██╗██╔══██╗██╔══██╗
|
|
██║ ███████║██║ █████╗ ██╔██╗ ██║██║ ██║███████║██████╔╝
|
|
██║ ██╔══██║██║ ██╔══╝ ██║╚██╗██║██║ ██║██╔══██║██╔══██╗
|
|
╚██████╗██║ ██║███████╗███████╗██║ ╚████║██████╔╝██║ ██║██║ ██║
|
|
╚═════╝╚═╝ ╚═╝╚══════╝╚══════╝╚═╝ ╚═══╝╚═════╝ ╚═╝ ╚═╝╚═╝ ╚═╝
|
|
|
|
status: {status}
|
|
|
|
"You are a stinky Poo"
|
|
- Ben de Roo, programmer
|
|
"""
|
|
|
|
print()
|
|
print()
|
|
print()
|
|
print("YOU CAN NOT START THE SERVER WITH PYTHON")
|
|
print("start with:")
|
|
print("$ bash start_server.sh")
|
|
print()
|
|
|