Spaces:
Sleeping
Sleeping
File size: 1,686 Bytes
c30d351 750073f c30d351 750073f c30d351 750073f c30d351 750073f c30d351 |
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 |
from fastapi import FastAPI, Response, WebSocket, WebSocketDisconnect, Depends, HTTPException
from fastapi.middleware.cors import CORSMiddleware
import uvicorn
from dotenv import load_dotenv
import os
# Load environment variables from .env file
load_dotenv()
IS_DEV = os.environ.get('ENV', 'DEV') != 'PROD'
SECURE_TOKEN = os.getenv("SECURE_TOKEN")
X_REQUEST_USER = os.environ.get('X_REQUEST_USER')
X_API_KEY = os.environ.get('X_API_KEY')
app = FastAPI()
# CORS Middleware: restrict access to only trusted origins
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
#allow_origins=["https://your-frontend-domain.com"],
#allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
# Simple token-based authentication dependency
async def authenticate_token(token: str):
if token != f"Bearer {SECURE_TOKEN}":
raise HTTPException(status_code=401, detail="Invalid token")
@app.get("/")
def root():
return Response(status=200, data='ok')
@app.get("/health")
def healthcheck():
return Response(status=200, data='ok')
# WebSocket endpoint
@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket, token: str = Depends(authenticate_token)):
await websocket.accept()
try:
while True:
data = await websocket.receive_text()
print(f"Message from client: {data}")
await websocket.send_text(f"Server received: {data}")
except WebSocketDisconnect:
print("Client disconnected")
def is_valid(u, p):
return u == X_REQUEST_USER and p == X_API_KEY
if __name__ == "__main__":
uvicorn.run('app:app', host='0.0.0.0', port=7860, reload=True) |