-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathlambda_function.py
More file actions
97 lines (77 loc) · 2.78 KB
/
Copy pathlambda_function.py
File metadata and controls
97 lines (77 loc) · 2.78 KB
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
import json
import boto3
import hashlib
import os
dynamodb = boto3.resource('dynamodb')
table_name = os.environ.get("TABLE_NAME", "url-shortener")
api_url = os.environ.get("API_GATEWAY_URL", "wlub7qavv0")
table = dynamodb.Table(table_name)
def generate_short_key(url):
return hashlib.sha256(url.encode()).hexdigest()[:6]
def lambda_handler(event, context):
print("Event:", json.dumps(event))
method = event.get("httpMethod")
path_params = event.get("pathParameters")
headers = {
"Access-Control-Allow-Origin": "*",
"Access-Control-Allow-Headers": "Content-Type",
"Access-Control-Allow-Methods": "OPTION,POST,GET",
"Content-Type": "application/json"
}
if method == "POST":
try:
body = json.loads(event.get("body", "{}"))
long_url = body.get("url")
if not long_url:
raise ValueError("Missing 'url' in request body")
short_key = generate_short_key(long_url)
# Save to DynamoDB
table.put_item(Item={
'short_key': short_key,
'long_url': long_url
})
short_url = f"https://{api_url}.execute-api.us-west-1.amazonaws.com/v0/{short_key}"
return {
"statusCode": 200,
"headers": headers,
"body": json.dumps({"shortUrl": short_url})
}
except Exception as e:
return {
"statusCode": 400,
"headers": headers,
"body": json.dumps({"message": str(e)})
}
elif method == "GET":
try:
short_key = path_params.get("short_code") if path_params else None
if not short_key:
raise ValueError("Missing short code in path")
response = table.get_item(Key={'short_key': short_key})
if 'Item' not in response:
return {
"statusCode": 404,
"headers": headers,
"body": json.dumps({"message": "Short code not found"})
}
long_url = response['Item']['long_url']
if not long_url.startswith("http"):
long_url = "https://" + long_url
return {
"statusCode": 302,
"headers": {
"Location": long_url # This tells the browser to redirect
},
"body": ""
}
except Exception as e:
return {
"statusCode": 500,
"headers": headers,
"body": json.dumps({"message": str(e)})
}
return {
"statusCode": 405,
"headers": headers,
"body": json.dumps({"message": f"Method {method} not allowed"})
}