ben/calendar
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

restructured

Browse Source
This commit is contained in:
ben
2026-08-17 15:50:41 +02:00
parent bbc8627867
commit 371b6364fe
5 changed files with 370 additions and 390 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/auth.py
+11 -1
View File
@@ -78,4 +78,14 @@ def get_current_user(
algorithms=[ALGORITHM]
)
return payload["user_id"]
return payload["user_id"]
def get_user(user_id):
return fetch_one(
"""
SELECT id, username
FROM users
WHERE id = ?
""",
(user_id,)
)
backend/database.py
+10
View File
@@ -73,3 +73,13 @@ if __name__ == "__main__":
print("-", table["name"])
else:
print("Geen tabellen gevonden.")
def get_table_names():
return fetch_all(
"""
SELECT name
FROM sqlite_master
WHERE type = 'table'
"""
)
backend/event_signups.py
+11
View File
@@ -52,4 +52,15 @@ def get_event_responses(event_id):
WHERE event_signups.event_id = ?
""",
(event_id,)
)
def get_event_responses(event_id):
return fetch_all(
"""
SELECT *
FROM event_signups
WHERE event_id = ?
""",
(event_id,)
)
backend/events.py
+43 -1
View File
@@ -173,4 +173,46 @@ def get_room_events(room_id):
ORDER BY start_time
""",
(room_id,)
)
)
def get_event(event_id):
return fetch_one(
"""
SELECT *
FROM events
WHERE id = ?
""",
(event_id,)
)
def get_event_room_id(event_id):
event = fetch_one(
"""
SELECT room_id
FROM events
WHERE id = ?
""",
(event_id,)
)
if event is None:
return None
return event["room_id"]
def get_event_creator(event_id):
event = fetch_one(
"""
SELECT creator_id
FROM events
WHERE id = ?
""",
(event_id,)
)
if event is None:
return None
return event["creator_id"]
backend/main.py
+295 -388
View File
@@ -1,30 +1,81 @@
from fastapi import FastAPI, Depends
from fastapi import FastAPI, Depends, HTTPException
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import PlainTextResponse
from pydantic import BaseModel
from typing import Literal
import sqlite3
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, get_room
from auth import (
create_user,
verify_user,
create_token,
get_current_user
)
from permissions import has_role
from events import create_event, get_user_events, check_event_threshold
from rooms import delete_room as delete_room_db
from rooms import change_room_name
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 rooms import (
create_room,
search_room,
accept_room,
get_rooms,
get_room,
get_room_members,
leave_room,
delete_room as delete_room_db,
change_room_name,
regenerate_invite_code
)
from events import (
create_event,
get_user_events,
check_event_threshold,
delete_event,
get_room_events,
update_event
)
from event_signups import (
respond_to_event,
count_confirmations
)
from database import get_table_names
from auth import (
create_user,
verify_user,
create_token,
get_current_user,
get_user
)
from events import (
create_event,
get_user_events,
check_event_threshold,
delete_event,
get_room_events,
update_event,
get_event,
get_event_room_id,
get_event_creator
)
from event_signups import (
respond_to_event,
count_confirmations,
get_event_responses
)
from events import delete_event, get_room_events
from rooms import regenerate_invite_code
class Login(BaseModel):
username: str
password: str
class GetRooms(BaseModel):
user_id: str
class UserCreate(BaseModel):
username: str
@@ -55,6 +106,11 @@ class RoomSearch(BaseModel):
class RoomJoin(BaseModel):
invite_code: str
class RoomUpdate(BaseModel):
new_name: str
class EventUpdate(BaseModel):
title: str | None = None
description: str | None = None
@@ -62,8 +118,6 @@ class EventUpdate(BaseModel):
end_time: str | None = None
min_people: int | None = None
class RoomUpdate(BaseModel):
new_name: str
app = FastAPI(
title="Calendar API",
@@ -71,14 +125,16 @@ app = FastAPI(
version="1.0"
)
app.add_middleware(
CORSMiddleware,
allow_origins=["*"], # later vervangen door je domein
allow_origins=["*"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
# general
@app.get("/")
def home():
@@ -90,16 +146,13 @@ def home():
@app.get("/database")
def database_status():
tables = fetch_all(
"SELECT name FROM sqlite_master WHERE type='table';"
)
tables = get_table_names()
return {
"tables": [table["name"] for table in tables]
}
# users
# user endpoints
@app.post("/create-user")
def api_create_user(user: UserCreate):
@@ -121,6 +174,7 @@ def api_create_user(user: UserCreate):
"user_id": user_id
}
@app.post("/login")
def api_login(login: Login):
@@ -147,54 +201,31 @@ def api_login(login: Login):
def get_me(
user_id: int = Depends(get_current_user)
):
user = get_user(user_id)
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)
# rooms
@app.delete("/rooms/{room_id}")
def delete_room(
room_id: int,
user_id: int = Depends(get_current_user)
@app.get("/users/{user_id}")
def api_get_user(
user_id: int,
current_user: int = Depends(get_current_user)
):
if not has_role(
room_id,
user_id,
["owner", "admin"]
):
return {
"status": "denied",
"reason": "no_permission"
}
user = get_user(user_id)
return delete_room_db(room_id)
if user is None:
raise HTTPException(
status_code=404,
detail="user not found"
)
@app.patch("/rooms/{room_id}")
def update_room(
room_id: int,
data: RoomUpdate,
user_id: int = Depends(get_current_user)
):
if not has_role(room_id, user_id, ["owner", "admin"]):
return {
"status": "denied",
"reason": "no_permission"
}
return change_room_name(
room_id,
data.new_name,
user_id
)
return dict(user)
# room endpoints
@app.post("/create-room")
def api_create_room(
@@ -203,7 +234,7 @@ def api_create_room(
):
try:
room_id = create_room(
result = create_room(
user_id,
room.room_name,
room.invite_code
@@ -217,13 +248,12 @@ def api_create_room(
return {
"status": "created",
"room_id": room_id
**result
}
@app.post("/search-room")
def api_search_room(
room: RoomSearch
):
def api_search_room(room: RoomSearch):
room_name = search_room(
room.invite_code
@@ -255,7 +285,7 @@ def api_join_room(
if result is None:
return {
"status": "failed",
"reason": "room_not_found"
"reason": "room_not_found_or_already_member"
}
return {
@@ -263,19 +293,151 @@ def api_join_room(
**result
}
@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 access"
)
return room
@app.get("/rooms/{room_id}/members")
def api_room_members(
room_id: int,
user_id: int = Depends(get_current_user)
):
if not has_role(room_id, user_id, "member"):
raise HTTPException(
status_code=403,
detail="room member permission required"
)
members = get_room_members(room_id)
return {
"members": [dict(member) for member in members]
}
@app.get("/rooms/{room_id}/events")
def api_room_events(
room_id: int,
user_id: int = Depends(get_current_user)
):
if not has_role(room_id, user_id, "member"):
raise HTTPException(
status_code=403,
detail="room member permission required"
)
events = get_room_events(room_id)
return {
"events": [
dict(event)
for event in events
]
}
# settings
@app.patch("/rooms/{room_id}")
def update_room(
room_id: int,
data: RoomUpdate,
user_id: int = Depends(get_current_user)
):
if not has_role(room_id, user_id, "admin"):
raise HTTPException(
status_code=403,
detail="admin permission required"
)
return change_room_name(
room_id,
data.new_name,
user_id
)
@app.post("/rooms/{room_id}/invite-code")
def update_invite_code(
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"
}
if not has_role(room_id, user_id, "admin"):
raise HTTPException(
status_code=403,
detail="admin permission required"
)
return regenerate_invite_code(room_id)
@app.delete("/rooms/{room_id}")
def delete_room(
room_id: int,
user_id: int = Depends(get_current_user)
):
if not has_role(room_id, user_id, "owner"):
raise HTTPException(
status_code=403,
detail="owner permission required"
)
return delete_room_db(room_id)
@app.delete("/rooms/{room_id}/members/me")
def api_leave_room(
room_id: int,
user_id: int = Depends(get_current_user)
):
if not has_role(room_id, user_id, "member"):
raise HTTPException(
status_code=403,
detail="not a room member"
)
leave_room(
room_id,
user_id
)
return {
"status": "left"
}
# events
@app.post("/events")
@@ -284,25 +446,11 @@ def api_create_event(
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 not has_role(event.room_id, user_id, "member"):
raise HTTPException(
status_code=403,
detail="room member permission required"
)
)
if member is None:
return {
"status": "failed",
"reason": "not_room_member"
}
event_id = create_event(
room_id=event.room_id,
@@ -335,21 +483,12 @@ def get_events(
}
@app.post("/events/{event_id}/respond")
def respond_event(
@app.get("/events/{event_id}")
def api_get_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,)
)
event = get_event(event_id)
if event is None:
raise HTTPException(
@@ -357,25 +496,33 @@ def respond_event(
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:
if not has_role(event["room_id"], user_id, "member"):
raise HTTPException(
status_code=403,
detail="not a room member"
detail="room member permission required"
)
return dict(event)
@app.post("/events/{event_id}/respond")
def respond_event(
event_id: int,
status: Literal["going", "maybe", "declined"],
user_id: int = Depends(get_current_user)
):
room_id = get_event_room_id(event_id)
if room_id is None:
raise HTTPException(
status_code=404,
detail="event not found"
)
if not has_role(room_id, user_id, "member"):
raise HTTPException(
status_code=403,
detail="room member permission required"
)
respond_to_event(
event_id,
@@ -390,198 +537,32 @@ def respond_event(
"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)
):
room_id = get_event_room_id(event_id)
event = fetch_one(
"""
SELECT room_id
FROM events
WHERE id = ?
""",
(event_id,)
)
if event is None:
if room_id 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:
if not has_role(room_id, user_id, "member"):
raise HTTPException(
status_code=403,
detail="no access"
detail="room member permission required"
)
responses = get_event_responses(event_id)
return {
"responses": [dict(r) for r in responses]
"responses": [dict(response) for response 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"
}
# more events
@app.patch("/events/{event_id}")
def api_update_event(
@@ -589,29 +570,20 @@ def api_update_event(
event: EventUpdate,
user_id: int = Depends(get_current_user)
):
creator_id = get_event_creator(event_id)
old = fetch_one(
"""
SELECT creator_id
FROM events
WHERE id = ?
""",
(event_id,)
)
if old is None:
if creator_id is None:
raise HTTPException(
status_code=404,
detail="event not found"
)
if old["creator_id"] != user_id:
if creator_id != user_id:
raise HTTPException(
status_code=403,
detail="not owner"
detail="event owner permission required"
)
update_event(
event_id,
event.title,
@@ -621,98 +593,42 @@ def api_update_event(
event.min_people
)
return {
"status": "updated"
}
@app.delete("/rooms/{room_id}/members/me")
def api_leave_room(
room_id: int,
@app.delete("/events/{event_id}")
def api_delete_event(
event_id: int,
user_id: int = Depends(get_current_user)
):
creator_id = get_event_creator(event_id)
member = fetch_one(
"""
SELECT *
FROM room_members
WHERE room_id = ?
AND user_id = ?
""",
(
room_id,
user_id
)
)
if member is None:
if creator_id is None:
raise HTTPException(
status_code=404,
detail="not member"
detail="event not found"
)
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:
if creator_id != user_id:
raise HTTPException(
status_code=403,
detail="no access"
detail="event owner permission required"
)
events = get_room_events(room_id)
delete_event(event_id)
return {
"events": [
dict(event)
for event in events
]
"status": "deleted"
}
from fastapi.responses import PlainTextResponse
import requests
# easter egg
@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"""
return """
██████╗ █████╗ ██╗ ███████╗███╗ ██╗██████╗ █████╗ ██████╗
██╔════╝██╔══██╗██║ ██╔════╝████╗ ██║██╔══██╗██╔══██╗██╔══██╗
██║ ███████║██║ █████╗ ██╔██╗ ██║██║ ██║███████║██████╔╝
@@ -720,17 +636,8 @@ def ascii_art():
╚██████╗██║ ██║███████╗███████╗██║ ╚████║██████╔╝██║ ██║██║ ██║
╚═════╝╚═╝ ╚═╝╚══════╝╚══════╝╚═╝ ╚═══╝╚═════╝ ╚═╝ ╚═╝╚═╝ ╚═╝
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()
status: online
"You are a stinky Poo"
- Ben de Roo, programmer
"""