createRooms.py

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)