-
Notifications
You must be signed in to change notification settings - Fork 0
/
Copy pathmain.py
144 lines (109 loc) · 3.96 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
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
import os
from enum import Enum
from typing import Union
from fastapi import FastAPI, Query, Request
from fastapi.responses import HTMLResponse
from slowapi import Limiter, _rate_limit_exceeded_handler
from slowapi.errors import RateLimitExceeded
from slowapi.util import get_remote_address
from starlette.responses import RedirectResponse
import config
import my_db
# pip install fastapi[all] gunicorn slowapi
# start test server:
# uvicorn main:app --reload --host 0.0.0.0 --port 8000
# start prod server:
# /srv/foo/bar/app/gunicorn -c gunicorn_conf.py main:app
root = os.path.dirname(os.path.abspath(__file__))
my_db.establish_db_connection()
base_url = os.getenv("FASTAPI_BASE_URL", "http://localhost:8000")
root_path = os.getenv("FASTAPI_ROOT_PATH", "/")
print(base_url, root_path)
limiter = Limiter(key_func=get_remote_address)
app = FastAPI(root_path=root_path)
app.state.limiter = limiter
app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler)
def get_status(request: Request):
"""
Currently hardcoded as `OK` but could easily report errors, number of active shortlinks, etc.
"""
return "OK"
@app.get("/", response_class=HTMLResponse)
async def root(request: Request):
"""
Respond with instructions.
"""
html_content = f"""
<html>
<head>
<title>URL shortener API using FastAPI</title>
</head>
<body>
<p>This is a URL shortener written in Python using FastAPI with SQLite3 for persistent storage. <a href="https://github.com/timoteostewart/fastapi-demo-url-shortener">Source code on GitHub.</a></p>
<p>Take it for a test drive using <a href="{root_path}docs">its OpenAPI interface</a>.</p>
<p>Alternatively, interact with it via `curl` from your terminal:</p>
<p>PowerShell:<br/>
curl -X 'POST' `<br/>
'{base_url}{root_path}?full_url=https://dev.thnr.net' `<br/>
-H 'accept: application/json'</p>
<p>*sh:<br/>
curl -X 'POST' \ <br/>
'{base_url}{root_path}?full_url=https://dev.thnr.net' \ <br/>
-H 'accept: application/json'</p>
</body>
</html>
"""
return HTMLResponse(content=html_content, status_code=200)
@app.get("/status/")
async def status(request: Request):
"""
Respond with current status (currently hardcoded as `OK`).
"""
return {"status": get_status()}
@app.get("/{short_url}")
async def redirect_from_short_url_to_long_url(request: Request, short_url):
"""
Redirect visitor to full_url, if short_url is valid.
"""
if not my_db.short_url_already_exists(short_url):
return {"error": "invalid short url"}
full_url = my_db.retrieve_full_url_from_short_url(short_url)
return RedirectResponse(url=full_url)
@app.get("/{short_url}/{admin_key}")
async def get_stats_for_short_url(request: Request, short_url=None, admin_key=None):
"""
Respond with the shortlink's statistics, if the `short_url` and `admin_key` are valid.
"""
try:
shortlink = my_db.retrieve_shortlink(short_url, admin_key)
return shortlink
except Exception as exc:
return {"error": str(exc)}
@app.post("/")
@limiter.limit("1/second")
async def create_short_url(
request: Request, full_url: str, short_url: Union[str, None] = None
):
"""
Given a `full_url`, create a shortlink and return its details
"""
try:
shortlink = my_db.create_short_url(full_url=full_url, short_url=short_url)
return shortlink
except Exception as exc:
return {"error": str(exc)}
@app.delete("/{short_url}/{admin_key}")
async def delete_short_url(request: Request, short_url=None, admin_key=None):
"""
Delete a shortlink from the database. This will free up its `short_url` for re-use.
"""
# check for valid `short_url` and `admin_key`
try:
shortlink = my_db.retrieve_shortlink(short_url, admin_key)
except Exception as exc:
return {"error": str(exc)}
try:
my_db.delete_short_url(short_url)
except Exception as exc:
return {"error": str(exc)}
return {"result": f"short_url `{short_url}` deleted"}