radames's picture
rooms api
c413f97
raw
history blame
3.44 kB
import os
from fastapi import Depends, FastAPI
import sqlite3
import requests
import uvicorn
from pathlib import Path
import json
app = FastAPI()
LIVEBLOCKS_SECRET = os.environ.get("LIVEBLOCKS_SECRET")
def get_db():
db = sqlite3.connect(Path("./rooms.db"), check_same_thread=False)
db.execute("CREATE TABLE IF NOT EXISTS rooms (id INTEGER PRIMARY KEY AUTOINCREMENT NOT NULL, room_id TEXT NOT NULL, users_count INTEGER NOT NULL DEFAULT 0)")
print("Connected to database")
db.commit()
db.row_factory = sqlite3.Row
try:
yield db
except Exception:
db.rollback()
finally:
db.close()
app = FastAPI()
rooms = ["sd-multiplayer-room-" + str(i) for i in range(0, 20)]
@app.get("/")
async def read_root(db: sqlite3.Connection = Depends(get_db)):
out = db.execute("SELECT * FROM rooms").fetchall()
print(out)
return out
@app.get("/create-rooms")
async def create_room(db: sqlite3.Connection = Depends(get_db)):
for room_id in rooms:
print(room_id)
createRoom(room_id, db)
all = db.execute("SELECT * FROM rooms").fetchall()
return all
def createRoom(room_id, db):
payload = {"id": room_id, "defaultAccesses": ["room:write"]}
response = requests.post(f"https://api.liveblocks.io/v2/rooms",
headers={"Authorization": f"Bearer {LIVEBLOCKS_SECRET}"}, json=payload)
# if response.status_code == 200:
data = response.json()
print(data)
if "error" in data and data["error"] == "ROOM_ALREADY_EXISTS":
print("Room already exists")
cursor = db.cursor()
cursor.execute("INSERT INTO rooms (room_id) VALUES (?)", (room_id,))
db.commit()
print("Room created")
print("Created room", room_id)
return True
def generateAuthToken():
response = requests.get(f"https://liveblocks.io/api/authorize",
headers={"Authorization": f"Bearer {LIVEBLOCKS_SECRET}"})
if response.status_code == 200:
data = response.json()
return data["token"]
else:
raise Exception(response.status_code, response.text)
def get_room_count(room_id: str, jwtToken: str = ''):
print("Getting room count" + room_id)
response = requests.get(
f"https://liveblocks.net/api/v1/room/{room_id}/users", headers={"Authorization": f"Bearer {jwtToken}", "Content-Type": "application/json"})
if response.status_code == 200:
res = response.json()
if "data" in res:
return len(res["data"])
else:
return 0
raise Exception("Error getting room count")
@app.get("/sync-rooms")
async def sync_rooms(db: sqlite3.Connection = Depends(get_db)):
try:
jwtToken = generateAuthToken()
rooms = db.execute("SELECT * FROM rooms").fetchall()
for row in rooms:
room_id = row["room_id"]
users_count = get_room_count(room_id, jwtToken)
print("Updating room", room_id, "with", users_count, "users")
cursor = db.cursor()
cursor.execute(
"UPDATE rooms SET users_count = ? WHERE room_id = ?", (users_count, room_id))
db.commit()
data = db.execute("SELECT * FROM rooms").fetchall()
return data
except Exception as e:
print(e)
return {"error": str(e)}
if __name__ == "__main__":
uvicorn.run("api:app", host="0.0.0.0", log_level="debug", reload=True)