Spaces:
Runtime error
Runtime error
File size: 3,517 Bytes
c413f97 8f150bf c413f97 8f150bf c413f97 8f150bf |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 |
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")
DB_PATH = Path("rooms.db")
if not DB_PATH.exists():
print("Creating database")
print("DB_PATH", DB_PATH)
db = sqlite3.connect(DB_PATH)
with open(Path("schema.sql"), "r") as f:
db.executescript(f.read())
db.commit()
db.close()
def get_db():
db = sqlite3.connect(Path("./rooms.db"), check_same_thread=False)
print("Connected to database")
db.row_factory = sqlite3.Row
try:
yield db
except Exception:
db.rollback()
finally:
db.close()
app = FastAPI()
rooms = ["room-" + str(i) for i in range(0, 41)]
@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("createRooms:app", host="0.0.0.0", log_level="debug", reload=True)
|