-
Notifications
You must be signed in to change notification settings - Fork 1
/
Copy pathmain.py
100 lines (78 loc) · 2.63 KB
/
main.py
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
99
# URL_Shortener/main.py
import secrets
import validators
from fastapi import Depends
from fastapi import FastAPI
from fastapi import HTTPException
from fastapi import Request
from fastapi.responses import RedirectResponse
from sqlalchemy.orm import Session
from starlette.datastructures import URL
from . import crud
from . import models
from . import schemas
from .config import get_settings
from .database import SessionLocal
from .database import engine
app = FastAPI()
models.Base.metadata.create_all(bind=engine)
def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()
def raise_bad_request(message):
raise HTTPException(status_code=400, detail=message)
def raise_not_found(request):
"""If the provided URL.key doesn't match any URLs
in our database, this function would be invoked.
"""
message = f"URL '{request.url}' doesn't exist"
raise HTTPException(status_code=404, detail=message)
@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.cread_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_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)
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