55 lines
1.7 KiB
Python
55 lines
1.7 KiB
Python
import time
|
|
from urllib.parse import urljoin
|
|
|
|
from fastapi import FastAPI, status, HTTPException, Depends
|
|
from fastapi.responses import RedirectResponse
|
|
|
|
from app.config import settings
|
|
from app.dao import URLsDAO
|
|
from app.migrations.shemas import SURL, SNewUrl, SGetStats
|
|
from app.r import get_redis_client
|
|
|
|
app = FastAPI(
|
|
title="Сокращатель ссылок",
|
|
)
|
|
|
|
|
|
@app.post("/zip_url", response_model=SNewUrl)
|
|
async def zip_url(url: SURL):
|
|
url = str(url.url)
|
|
|
|
url_hash = await URLsDAO.add_new_url(original_url=url)
|
|
|
|
new_url = urljoin(settings.SHORTENER_HOST, str(url_hash))
|
|
|
|
return {"new_url": new_url}
|
|
|
|
|
|
@app.get("/{url_hash}", response_class=RedirectResponse, status_code=status.HTTP_307_TEMPORARY_REDIRECT)
|
|
async def redirect(url_hash: int, r=Depends(get_redis_client)):
|
|
if r.exists(str(url_hash)):
|
|
url_bytes = r.get(str(url_hash))
|
|
url = url_bytes.decode('utf-8')
|
|
else:
|
|
urls = await URLsDAO.find_by_hash(url_hash)
|
|
if not urls:
|
|
raise HTTPException(status_code=404, detail="Этого адреса не существует")
|
|
url = urls.original_url
|
|
r.set(str(url_hash), url)
|
|
r.expire(str(url_hash), 60)
|
|
|
|
list_key = f"visit_times:{url_hash}"
|
|
timestamp = int(time.time())
|
|
r.lpush(list_key, timestamp)
|
|
|
|
return RedirectResponse(url=url)
|
|
|
|
|
|
@app.get("/stats/{url_hash}", response_model=SGetStats)
|
|
async def get_stats(url_hash: int, r=Depends(get_redis_client)):
|
|
list_key = f"visit_times:{url_hash}"
|
|
if r.exists(list_key):
|
|
timestamps = r.lrange(list_key, 0, -1)
|
|
visit_times = len(timestamps)
|
|
return {"visit_times": visit_times, "timestamps": timestamps}
|
|
return {"visit_times": 0, "timestamps": []}
|