-
Notifications
You must be signed in to change notification settings - Fork 5.3k
Expand file tree
/
Copy pathmain.py
More file actions
94 lines (72 loc) · 2.49 KB
/
main.py
File metadata and controls
94 lines (72 loc) · 2.49 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
import validators
from sqlalchemy.orm import Session
from starlette.datastructures import URL
from fastapi import Depends, FastAPI, HTTPException, Request
from fastapi.responses import RedirectResponse
from . import crud, models, schemas
from .config import get_settings
from .database import SessionLocal, engine
app = FastAPI()
models.Base.metadata.create_all(bind=engine)
def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()
def get_admin_info(db_url: models.URL) -> schemas.URLInfo:
base_url = URL(get_settings().base_url)
admin_endpoint = app.url_path_for(
"administration info", secret_key=db_url.secret_key
)
db_url.url = str(base_url.replace(path=db_url.key))
db_url.admin_url = str(base_url.replace(path=admin_endpoint))
return db_url
def raise_bad_request(message):
raise HTTPException(status_code=400, detail=message)
def raise_not_found(request):
message = f"URL '{request.url}' doesn't exist"
raise HTTPException(status_code=404, detail=message)
@app.get("/")
def read_root():
return "Welcome to the URL shortener API :)"
@app.post("/url", response_model=schemas.URLInfo)
def create_url(url: schemas.URLBase, db: Session = Depends(get_db)):
if not validators.url(url.target_url):
raise_bad_request(message="Your provided URL is not valid")
db_url = crud.create_db_url(db=db, url=url)
return get_admin_info(db_url)
@app.get("/{url_key}")
def forward_to_target_url(
url_key: str, request: Request, db: Session = Depends(get_db)
):
if db_url := crud.get_db_url_by_key(db=db, url_key=url_key):
crud.update_db_clicks(db=db, db_url=db_url)
return RedirectResponse(db_url.target_url)
else:
raise_not_found(request)
@app.get(
"/admin/{secret_key}",
name="administration info",
response_model=schemas.URLInfo,
)
def get_url_info(
secret_key: str, request: Request, db: Session = Depends(get_db)
):
if db_url := crud.get_db_url_by_secret_key(db, secret_key=secret_key):
return get_admin_info(db_url)
else:
raise_not_found(request)
@app.delete("/admin/{secret_key}")
def delete_url(
secret_key: str, request: Request, db: Session = Depends(get_db)
):
if db_url := crud.deactivate_db_url_by_secret_key(
db, secret_key=secret_key
):
message = (
f"Successfully deleted shortened URL for '{db_url.target_url}'"
)
return {"detail": message}
else:
raise_not_found(request)