72 lines
2.2 KiB
Python
72 lines
2.2 KiB
Python
from fastapi import FastAPI, HTTPException
|
|
from fastapi.responses import RedirectResponse, FileResponse
|
|
from fastapi.staticfiles import StaticFiles
|
|
from dotenv import load_dotenv
|
|
import couchdb
|
|
import os
|
|
import uuid
|
|
import validators
|
|
|
|
load_dotenv()
|
|
|
|
COUCHDB_USER = os.getenv("COUCHDB_USER")
|
|
COUCHDB_PASSWORD = os.getenv("COUCHDB_PASSWORD")
|
|
COUCHDB_HOST = os.getenv("COUCHDB_HOST")
|
|
COUCHDB_PORT = os.getenv("COUCHDB_PORT")
|
|
|
|
app = FastAPI()
|
|
app.mount("/static", StaticFiles(directory="static", html=True), name="static")
|
|
|
|
couch = couchdb.Server(f"http://{COUCHDB_USER}:{COUCHDB_PASSWORD}@{COUCHDB_HOST}:{COUCHDB_PORT}")
|
|
|
|
@app.get("/")
|
|
async def root():
|
|
return FileResponse('static/index.html')
|
|
|
|
@app.get("/{url_id}", status_code=301, response_class=RedirectResponse)
|
|
async def redirect_urls(url_id):
|
|
db = couch["urls"]
|
|
if url_id in db:
|
|
target_url = db[url_id].get('full_url')
|
|
return RedirectResponse(target_url)
|
|
else:
|
|
return RedirectResponse('static/404.html')
|
|
|
|
|
|
# TODO: Get JWT tokens working and use that to return only the user's urls
|
|
# TODO: Look into how FastAPI handles the Authentication header
|
|
@app.get("/api/v1/urls")
|
|
async def read_urls():
|
|
db = couch["urls"]
|
|
return [ { id: db[id].get('full_url') } for id in db ]
|
|
|
|
@app.get("/api/v1/urls/{url_id}")
|
|
async def read_url(url_id):
|
|
db = couch["urls"]
|
|
if url_id in db:
|
|
return db[url_id]
|
|
else:
|
|
raise HTTPException(status_code=404, detail="Url not found")
|
|
|
|
# TODO: Throttle
|
|
# TODO: If user not found, generate a new JWT
|
|
@app.put("/api/v1/urls")
|
|
async def create_url(body: dict):
|
|
if validators.url(body["url"]):
|
|
db = couch["urls"]
|
|
# TODO: Calculate how many unique IDs we are actually generating
|
|
url_id = uuid.uuid4().hex[:6]
|
|
db[url_id] = { "full_url": body["url"], "user_id": body["username"] }
|
|
return { "shortenedUrl": "https://zipmy.link/" + url_id }
|
|
else:
|
|
raise HTTPException(status_code=400, detail="Url provided is invalid")
|
|
|
|
@app.delete("/api/v1/urls/{url_id}")
|
|
async def delete_url(url_id):
|
|
db = couch["urls"]
|
|
if url_id in db:
|
|
del db[url_id]
|
|
return { "message": "Url deleted" }
|
|
else:
|
|
raise HTTPException(status_code=404, detail="Url not found")
|