57 lines
1.8 KiB
Python
57 lines
1.8 KiB
Python
import os
|
|
|
|
from sqlalchemy.exc import IntegrityError
|
|
from fastapi import FastAPI, HTTPException
|
|
from fastapi.responses import RedirectResponse
|
|
|
|
from dependencies.models import UrlMapping
|
|
from dependencies.database import database, Base, engine
|
|
from schemas import UrlPayload
|
|
from utils import generate_shortcode
|
|
|
|
app = FastAPI()
|
|
|
|
@app.on_event("startup")
|
|
async def startup():
|
|
await database.connect()
|
|
Base.metadata.create_all(bind=engine) # TODO: create all tables if not present (ONLY DEVELOPMENT)
|
|
|
|
@app.on_event("shutdown")
|
|
async def shutdown():
|
|
await database.disconnect()
|
|
|
|
|
|
|
|
@app.get("/{shortcode}")
|
|
async def redirect_to_original_url(shortcode: str):
|
|
# TODO: do validation check for valid shortcode
|
|
if not shortcode:
|
|
raise HTTPException(status_code=400, detail="Shortcode is not valid")
|
|
|
|
query = UrlMapping.__table__.select().where(UrlMapping.shortcode == shortcode)
|
|
url_mapping = await database.fetch_one(query)
|
|
|
|
if url_mapping:
|
|
return RedirectResponse(url=url_mapping.url, status_code=302)
|
|
else:
|
|
raise HTTPException(status_code=404, detail="Shortcode not found")
|
|
|
|
@app.post("/shorten")
|
|
async def create_url_shortcode(url_payload: UrlPayload):
|
|
original_url = str(url_payload.url) # convert to string from HttpUrl
|
|
shortcode = generate_shortcode()
|
|
|
|
try:
|
|
query = UrlMapping.__table__.insert().values({
|
|
"url": original_url,
|
|
"shortcode": shortcode
|
|
})
|
|
await database.execute(query)
|
|
|
|
return {"shortcode": shortcode, "original_url": original_url}
|
|
except IntegrityError as e:
|
|
if 'shortcode' in str(e):
|
|
raise HTTPException(status_code=409, detail="Shortcode already in use")
|
|
else:
|
|
raise HTTPException(status_code=500, detail="Internal Server Error")
|
|
|