Code Examples

from xyra import App, Request, Response

app = App()

@app.get("/")
def home(req: Request, res: Response):
    res.json({"message": "Hello, World!"})

if __name__ == "__main__":
    app.listen(8000, logger=True)

from xyra import App, Request, Response

app = App()

# In-memory "database"
todos = []
next_id = 1

@app.get("/todos")
def get_todos(req: Request, res: Response):
    res.json(todos)

@app.post("/todos")
async def create_todo(req: Request, res: Response):
    global next_id
    data = await req.json()
    if not data or "task" not in data:
        res.status(400).json({"error": "Task is required"})
        return
    
    new_todo = {"id": next_id, "task": data["task"], "completed": False}
    todos.append(new_todo)
    next_id += 1
    res.status(201).json(new_todo)

@app.get("/todos/{todo_id}")
def get_todo(req: Request, res: Response):
    todo_id = int(req.params.get("todo_id"))
    todo = next((t for t in todos if t["id"] == todo_id), None)
    
    if todo:
        res.json(todo)
    else:
        res.status(404).json({"error": "Todo not found"})

if __name__ == "__main__":
    app.listen(8000, logger=True)

from xyra import App, WebSocket

app = App()

# Note: For production, a more robust solution like Redis Pub/Sub is recommended
connections = set()

async def on_open(ws: WebSocket):
    print("Client connected.")
    connections.add(ws)
    await ws.send("Welcome to the chat!")

async def on_message(ws: WebSocket, message: str, opcode: int):
    # Broadcast the message to all other connected clients
    for client in connections:
        if client != ws:
            await client.send(f"User says: {message}")

async def on_close(ws: WebSocket, code: int, message: str | None):
    print("Client disconnected.")
    connections.discard(ws)

app.websocket("/ws", {
    "open": on_open,
    "message": on_message,
    "close": on_close
})

if __name__ == "__main__":
    app.listen(8000, logger=True)