|
|
import inspect |
|
|
from functools import wraps |
|
|
from typing import get_type_hints |
|
|
|
|
|
|
|
|
def strict_types(func): |
|
|
sig = inspect.signature(func) |
|
|
type_hints = get_type_hints(func) |
|
|
|
|
|
@wraps(func) |
|
|
def wrapper(*args, **kwargs): |
|
|
bound_args = sig.bind(*args, **kwargs) |
|
|
bound_args.apply_defaults() |
|
|
|
|
|
|
|
|
for name, value in bound_args.arguments.items(): |
|
|
expected_type = type_hints.get(name) |
|
|
if expected_type and not isinstance(value, expected_type): |
|
|
raise TypeError( |
|
|
f"Argument '{name}' expected { |
|
|
expected_type.__name__}, got { |
|
|
type(value).__name__}") |
|
|
|
|
|
|
|
|
result = func(*args, **kwargs) |
|
|
|
|
|
|
|
|
expected_return = type_hints.get('return') |
|
|
if expected_return and not isinstance(result, expected_return): |
|
|
raise TypeError( |
|
|
f"Return value expected { |
|
|
expected_return.__name__}, got { |
|
|
type(result).__name__}") |
|
|
|
|
|
return result |
|
|
|
|
|
return wrapper |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
from fastapi import FastAPI, HTTPException |
|
|
from pydantic import BaseModel, conint |
|
|
import threading |
|
|
import time |
|
|
|
|
|
app = FastAPI() |
|
|
live_names = {} |
|
|
expiration_times = {} |
|
|
|
|
|
class LiveNameRequest(BaseModel) : |
|
|
name: str |
|
|
url: str |
|
|
timeout: conint(ge=60, le=3600) |
|
|
|
|
|
@strict_types |
|
|
def remove_expired_name(name: str) : |
|
|
time.sleep(expiration_times[name] - time.time()) |
|
|
if name in live_names: |
|
|
del live_names[name] |
|
|
del expiration_times[name] |
|
|
print(f"{name} has expired and is available again!") |
|
|
|
|
|
@app.post("/set_live_name") |
|
|
def set_live_name(data: LiveNameRequest) : |
|
|
if data.name in live_names : |
|
|
remaining_time = max(0, int((expiration_times[data.name] - time.time()) / 60)) |
|
|
return {"message": f"Name taken. Available in {remaining_time} min(s)."} |
|
|
|
|
|
live_names[data.name] = data.url |
|
|
expiration_times[data.name] = time.time() + data.timeout |
|
|
|
|
|
threading.Thread(target=remove_expired_name, args=(data.name,)).start() |
|
|
|
|
|
return {"message": f"Live name set successfully (expires in {data.timeout//60} min)", |
|
|
"access_url": f"/lookup/{data.name}"} |
|
|
|
|
|
@app.get("/lookup/name") |
|
|
def lookup_name(c: str) : |
|
|
if c in live_names : |
|
|
return {"url": live_names[c]} |
|
|
raise HTTPException(status_code=404, detail="Live name not found") |