-
Notifications
You must be signed in to change notification settings - Fork 1
Expand file tree
/
Copy pathmain.py
More file actions
98 lines (70 loc) · 2.6 KB
/
main.py
File metadata and controls
98 lines (70 loc) · 2.6 KB
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
from fastapi import FastAPI, HTTPException, Request
from pydantic import BaseModel
from models.links import Link
from tools.validations import link_validation
from tools.utils import create_short_link_record, check_link_is_exists, set_cache
from starlette.responses import RedirectResponse
from tools.database import redis_obj as redis
from fastapi.templating import Jinja2Templates
from fastapi.staticfiles import StaticFiles
app = FastAPI()
templates = Jinja2Templates(directory="templates")
app.mount("/static", StaticFiles(directory="static"), name="static")
class LinkData(BaseModel):
address: str
@app.post("/generate")
async def generate(link: LinkData, request: Request):
"""
This view is responsible for receiving links and creating short links
params : link : user input link for shorting
params : request : http request object
return : short link in json or HTTPException
"""
link_address = link.address
cache = redis.get(link_address)
if cache:
return {"link": f"{request.client.host}/{cache.decode()}"}
is_exists = check_link_is_exists(link_address)
if is_exists:
set_cache(link_address, is_exists, 60)
return {"link": f"{request.client.host}/{is_exists}"}
link_validation(link_address)
random_link = create_short_link_record(link_address)
set_cache(link_address, random_link, 60)
return {"link": f"{request.client.host}/{random_link}"}
@app.get("/{link}")
async def redirect(link: str):
"""
This function is responsible for checking the short link and
redirect users to the path related to that short link
params : short_link : user short link
return : redirect user or HTTPException
"""
redirect_link = Link.select().where(Link.short_link == link)
if redirect_link.exists():
return RedirectResponse(url=redirect_link[0].address)
else:
raise HTTPException(status_code=400, detail="Short url doesn't exist")
@app.get("/")
async def root(request: Request):
"""
Web site index
"""
return templates.TemplateResponse("index.html", {"request": request})
@app.get("/admin/")
async def admin(request: Request):
"""
admin panel
"""
total_urls_cache = redis.dbsize()
last_10_link = Link.select().limit(10)
count_link_on_db = Link.select().count()
print(last_10_link)
print(total_urls_cache)
context = {
"request": request,
"total_urls_cache" : total_urls_cache,
"last_10_link" : last_10_link,
"count_link_on_db": count_link_on_db,
}
return templates.TemplateResponse("admin/index.html", context)