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
|
from fastapi import APIRouter, status, Path, Depends
from fastapi.exception_handlers import HTTPException
from typing import Annotated
import string
import random
import datetime
import validators
from app.util.db_dependency import get_db
from models import Link, Log
from app.schemas.links_schemas import URLSchema
from app.schemas.auth_schemas import User
from app.util.authentication import get_current_user
router = APIRouter(prefix="/links", tags=["links"])
@router.get("/", summary="Get all of the links associated with your account")
async def get_links(
current_user: Annotated[User, Depends(get_current_user)],
db=Depends(get_db),
):
links = db.query(Link).filter(Link.owner == current_user.id).all()
if not links:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND, detail="No links found"
)
return links
@router.post("/", summary="Create a new link")
async def create_link(
url: URLSchema,
current_user: Annotated[User, Depends(get_current_user)],
db=Depends(get_db),
):
# Check if the URL is valid
if not validators.url(url.url):
raise HTTPException(
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
detail="Invalid URL",
)
# Create the new link and add it to the database
while True:
try:
link_path = "".join(
random.choices(string.ascii_uppercase + "1234567890", k=5)
).upper()
new_link = Link(
link=link_path,
owner=current_user.id,
redirect_link=url.url,
)
db.add(new_link)
db.commit()
break
except:
continue
return {"link": link_path, "expire_date": new_link.expire_date}
@router.delete("/{link}", summary="Delete a link")
async def delete_link(
link: Annotated[str, Path(title="Link to delete")],
current_user: Annotated[User, Depends(get_current_user)],
db=Depends(get_db),
):
"""
Delete a link and all of the logs associated with it
"""
link = link.upper()
# Get the link and check the owner
link = db.query(Link).filter(Link.link == link).first()
if not link:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND, detail="Link not found"
)
if link.owner != current_user.id:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Link not associated with your account",
)
# Get and delete all logsk
logs = db.query(Log).filter(Log.link == link.link).all()
for log in logs:
db.delete(log)
# Delete the link
db.delete(link)
db.commit()
return status.HTTP_204_NO_CONTENT
@router.get("/{link}/logs", summary="Get all logs associated with a link")
async def get_link_logs(
link: Annotated[str, Path(title="Link to get logs for")],
current_user: Annotated[User, Depends(get_current_user)],
db=Depends(get_db),
):
"""
Get all of the IP logs associated with a link
"""
link = link.upper()
# Get the link and check the owner
link = db.query(Link).filter(Link.link == link).first()
if not link:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND, detail="Link not found"
)
if link.owner != current_user.id:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Link not associated with your account",
)
# Get and return all of the logs - ordered by timestamp
logs = (
db.query(Log)
.filter(Log.link == link.link)
.order_by(Log.timestamp.desc())
.all()
)
return logs
@router.delete("/{link}/logs", summary="Delete logs associated with a link")
async def delete_link_logs(
link: Annotated[str, Path(title="Link to delete logs for")],
current_user: Annotated[User, Depends(get_current_user)],
db=Depends(get_db),
):
"""
Delete all of the IP logs associated with a link
"""
link = link.upper()
# Get the link and check the owner
link = db.query(Link).filter(Link.link == link).first()
if not link:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND, detail="Link not found"
)
if link.owner != current_user.id:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Link not associated with your account",
)
# Get all of the logs
logs = db.query(Log).filter(Log.link == link.link).all()
for log in logs:
db.delete(log)
db.commit()
return status.HTTP_204_NO_CONTENT
|
