-
Notifications
You must be signed in to change notification settings - Fork 0
/
main.py
196 lines (171 loc) · 7.06 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
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
# import qrcode
import base64
import os
import json
import traceback
import shortuuid
# FastAPI
from fastapi import Depends, FastAPI, HTTPException, Request, status
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import HTMLResponse, JSONResponse, RedirectResponse
from fastapi.templating import Jinja2Templates
from fastapi.encoders import jsonable_encoder
# instagram
from instaloader import Instaloader, Profile
from instaloader.exceptions import ProfileNotExistsException
# # other
from common.config import COLLECTION_NAME, DB_NAME, MONGO_DB_URL
from common.constants import SLUG_LIMIT, USER_ID
from db.db_operations import UrlShortenerDB
from schemas.urlShortener import UrlSchema, UrlUpdateSchema, TestSchema
from utils.helper_utilities import generate_qr_code
# # ******************----------- End Of Imports -----------******************
app = FastAPI()
# Templates
templates = Jinja2Templates(directory="templates")
# Create an instance of the database wrapper class
url_shortner_db = UrlShortenerDB(
db_url=MONGO_DB_URL, db_name=DB_NAME, collection_name=COLLECTION_NAME
)
# CORS middleware
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
# # FastApi End points
@app.get("/", response_class=HTMLResponse)
async def index(request: Request):
# print(f"{(json.dumps(dict(request),default=str, indent=4))}")
# print(f"{request.client}")
return templates.TemplateResponse("index.html", {"request": request})
@app.post("/api/shorten", response_model=dict)
async def create_short_url(url: UrlSchema):
response = {}
try:
# convert pydantic schema object to python dict
url = dict(url)
slug = url.get("customSlugCode") if url.get('customSlugCode') else shortuuid.ShortUUID().random(length=SLUG_LIMIT)
# check is url exist in DB , IF so raise an exception
db_url_data = url_shortner_db.get_url_data_by_slug(slug=slug)
if db_url_data:
raise HTTPException(status_code=404, detail=f"Slug code is invalid, It has been used. found {db_url_data}")
url_data = {
"userID": USER_ID,
"longUrl": url.get("longUrl"),
"slug": slug,
}
# if url.get('makeQrcode'):
# response['qr_code'] = generate_qr_code(url=url.get("longUrl"))
inserted = url_shortner_db.create_url(data=url_data)
response = {
'message': f"create short url for url={url['longUrl']}",
'data': url_data,
'inserted': inserted,
}
return JSONResponse(response, 200)
except HTTPException as http_exception:
response['error'] = http_exception.detail
return JSONResponse(response, http_exception.status_code)
except Exception as err:
response[
"error"
] = f"An Error occurred. Error: {err}"
response['traceback'] = traceback.format_exc()
return JSONResponse(response, 500)
@app.get("/api/get/slugs/{user_id}")
async def get_slugs_for_user(user_id: str) -> list:
"""Get slugs for a particular user."""
response = {}
try:
urls_data = url_shortner_db.get_urls_data_by_user_id(user_id=user_id)
if len(urls_data) > 0:
response["data"] = urls_data
else:
response["message"] = f"no data found for a user {user_id}"
return JSONResponse(response, 200)
except Exception as err:
response["error"] = f"Error: {err}."
response['traceback'] = traceback.format_exc()
return JSONResponse(response, 500)
@app.put("/api/update/{user_id}/slug/{slug_id}")
async def get_slug_data(user_id: str, slug_id: str, update_url: UrlUpdateSchema):
response = {}
long_url = update_url.longUrl
try:
result = url_shortner_db.update_url(user_id=user_id, slug=slug_id, long_url=long_url)
response['message'] = (
f'Successfully update the URL={long_url} for User: {user_id} & Slug:{slug_id}.'
if result
else f"Provided Slug={slug_id} does not exist in DB."
)
return JSONResponse(response, 200)
except Exception as err:
response['error'] = err
response['traceback'] = traceback.format_exc()
return JSONResponse(response, 500)
# # API redirection for Slug
@app.get("/{slug}", response_class=RedirectResponse)
async def redirect_slug(slug: str):
"""Redirect of slug url using Long url."""
response = {}
try:
slug_data = url_shortner_db.get_url_data_by_slug(slug=slug)
if not slug_data:
raise HTTPException(status_code=404, detail="URL not found !")
url_shortner_db.increment_click_count(user_id=slug_data['userID'], slug=slug_data['slug'])
return slug_data.get("longUrl")
except HTTPException as err:
response["error"] = f'url not found for parameter: {slug}'
return JSONResponse(response, 404)
except Exception as err:
response["error"] = f"{err}"
response['traceback'] = traceback.format_exc()
return JSONResponse(response, 500)
@app.post("/api/test/{method}")
async def test(method: str, test_schema: TestSchema, request: Request):
response = {}
try:
if method == "get":
response = {
"get_urls_data_by_user_id": url_shortner_db.get_urls_data_by_user_id(
user_id=test_schema.userID
),
"get_data_by_user_and_slug": url_shortner_db.get_data_by_user_and_slug(
user_id=test_schema.userID, slug= test_schema.slugCode or "Gamil"
),
"get_url_data_by_slug": url_shortner_db.get_url_data_by_slug(slug= test_schema.slugCode or "MyGithub"),
'request': request.client.host,
}
elif method == 'deleteMany':
data = {
'userID': test_schema.userID or 'CHETAN_JARANDE',
'longUrl': test_schema.longUrl or 'https://github.com/Chetan_Jarande31',
}
response = {
'delete_many': url_shortner_db.delete_many_url(filter_data=data),
'filter_data': data,
}
else:
response["error"] = f'provided method={method} is invalid.'
return JSONResponse(response, 200)
except Exception as err:
response["error"] = f"error : {err}."
response['traceback'] = traceback.format_exc()
return JSONResponse(response, 500)
# # *****************------------------------ Instagram Loader ------------------------*****************
# insta profile picture url
@app.get("/api/instagram/get-insta-profile/{profile_id}")
async def get_insta_profile_url(profile_id: str):
response= {}
try:
response = {
"instagramProfilePicUrl": Profile.from_username(Instaloader().context, profile_id).profile_pic_url
}
except ProfileNotExistsException as insta_profile_not_exist_exception:
response['error'] = str(insta_profile_not_exist_exception)
except Exception as err:
response['error'] = err
return response