an aws s3 storage blog with simple lambda api supports posts and tags

s3blogapi.py

#!/usr/bin/env python3

import os, re, sys, time, uuid, json, pprint, pathlib, collections, boto3

base_path = "basedir"
bucket_id = "yourbucketname"

#if not hasattr(re, "Match"):
#	import _sre
#	re.Match = _sre.SRE_Match

class S3(object):

	gen_uuid = uuid.uuid1
	resource = boto3.resource("s3")
	client = boto3.client("s3")
	bucket = lambda b=bucket_id: S3.resource.Bucket(b)
	obj = lambda n, b=bucket_id: S3.resource.Object(b, n)

"""

# ? List Schema: {
	id(unixtime): {
		title: str,
		uuid: str(uuid),
		tags: [ str, ... ],
	}
	...
}

# Post Schema: {
	id: int(autoassign?),
	title: str,
	tags: [ str, ... ],
	date: int(unixtime),
	edit: int(unixtime),
	attachments: { name: url | base64(data) },
	desc: str(short desc #search),
	body: str(markdown),
	...
}
- / (root):
		GET:    . [ id(str), ... ]
		PUT:    . -
		POST:   . { title: str, ... } -> id(str)
		PATCH:  . -
		DELETE: . -
	- /{item}:
		GET:    . { #postschema, }
		PUT:    .
		POST:   .
		PATCH:  .
		DELETE: .

# Tag Schema: {
	name: [ id(post), ... ], # hex?
	...
}
- methods:
	- / (root):
		HEAD:   . # n-of-posts
		GET:    . [ tag(str), ... ]
		PUT:    . -
		POST:   . [ tag(str), ... ] # create given tags
		PATCH:  . -
		DELETE: . -
	- /{item}:
		GET:    . [ str(post#id), ... ]
		PUT:    . # replace contents, set directly
		POST:   . [ add post id to tag ]
		PATCH:  . [ str(post#id), ... ] -> remove posts
		DELETE: . # id(str)

"""

class DotDict(collections.OrderedDict):

	def __getattr__(self, key, *args):
		return self.__getitem__(key)

	def __setattr__(self, key, val, *args):
		return self.__setitem__(key, val)

	pass


class ApiBase(DotDict):

	path = os.path.join(base_path, "test")

	def __init__(self, *args):
		return #print(__class__, self, args, self.routes)

	def __call__(self, e: dict, m):
		return helper_apicall(e, m, self.routes)

	routes = DotDict({
		"^([/]?)?$": (200, {}, {
			"use": [ "fn", "str", "list", "dict", ]
		}),
		"^([/]?(fn)[/]?)$": lambda e, m: (201, {}, "function is ok"),
		"^([/]?(str)[/]?)$":  (201, {}, "string is ok"),
		"^([/]?(list)[/]?)$": (201, {}, [ "list", "is", "ok" ]),
		"^([/]?(dict)[/]?)$": (201, {}, dict(dict="okay")),
	})

	pass


class ApiPosts(ApiBase):

	path = os.path.join(base_path, "posts")
	post = lambda name, ext="json": os.path.join(
		ApiPosts.path, ".".join([ name, ext ])
	)
	load = lambda name: json.load(S3.obj(ApiPosts.post(name)).get()["Body"])
	save = lambda name, data: S3.obj(ApiPosts.post(name)).put(
		ACL="public-read", Body=json.dumps(data).encode()
	)
	getk = lambda e: (" ".join(re.findall("[0-9A-Fa-f]+", e["path"])))

	def __init__(self, *args):
		return #print(__class__, self, args, self.routes)

	def __call__(self, e: dict, m):
		return helper_apicall(e, m, self.routes)

	@staticmethod
	def safe(arg):
		try:
			return json.load(arg)
		except Exception as e:
			return str(e)
			pass
		return None

	@staticmethod
	def listall(e, m):
		try:
			data = S3.client.list_objects(
				Prefix = str(ApiPosts.path),
				Bucket = bucket_id,
			)
			data = list(map(lambda c: {
				**c, "LastModified": str(c["LastModified"])
			}, sorted(
				data["Contents"], reverse=True,
				key=lambda p: p["LastModified"],
			)))
			filtered = list(filter(lambda m: (
				"Key" in m.keys() and m["Key"].endswith(".json")
			), data))
			result = list(map(lambda p: {
				pathlib.Path(p["Key"]).name.split(".")[0]: (
					ApiPosts.safe(S3.obj(p["Key"]).get()["Body"])
				)
			}, filtered))
			final = list(map(lambda x: {
				**ApiPosts.safe(S3.obj(x["Key"]).get()["Body"]),
				"id": pathlib.Path(x["Key"]).name.split(".")[0],
			}, filtered))
			return (200, { "content-type": "application/json" }, final)
		except Exception as err:
			return (400, {}, { "error": str(err), })
			pass
		return (400, {}, None)

	@staticmethod
	def create(e, m):
		try:
			body = json.loads(e["body"])
			if not isinstance(body, dict):
				raise Exception("body must be json-object!")
			ident = S3.gen_uuid().hex
			obj = S3.obj(ApiPosts.post(ident))
			obj.put(ACL="public-read", Body=json.dumps(body).encode())
			return (200, {}, { "id": ident, "key": obj.key, "data": body, })
		except Exception as err:
			return (400, {}, { "error": str(err), })
			pass
		return (400, {}, None)

	@staticmethod
	def delete(e, m):
		try:
			ident = ApiPosts.getk(e)
			obj = S3.obj(ApiPosts.post(ident))
			return (200, {}, { "result": obj.delete(), })
		except Exception as err:
			return (400, {}, { "error": str(err), })
			pass
		return (200, {}, None)

	@staticmethod
	def update(e, m):
		try:
			body = json.loads(e["body"])
			if not isinstance(body, dict):
				raise Exception("body must be json-object!")
			ident = ApiPosts.getk(e)
			obj = S3.obj(ApiPosts.post(ident))
			data = {
				**(json.load(obj.get()["Body"])),
				**body,
			}
			obj.put(ACL="public-read", Body=json.dumps(data).encode())
			return (200, {}, data)
		except Exception as err:
			return (400, {}, { "error": str(err), })
			pass
		return (400, {}, None)

	@staticmethod
	def get(e, m):
		try:
			ident = ApiPosts.getk(e)
			obj = S3.obj(ApiPosts.post(ident))
			data = json.load(obj.get()["Body"])
			return (200, {}, data)
		except Exception as err:
			return (400, {}, { "error": str(err), })
			pass
		return (400, {}, None)

	routes = DotDict({
		"^[/]?$": DotDict({
			"DEF": "refer to docs",
			"GET": listall.__func__,
			"POST": create.__func__,
		}),
		"^/[0-9A-Fa-f]+$": DotDict({
			"DEF": "refer to docs",
			"GET": get.__func__,
			"PUT": update.__func__,
			"POST": "i didn't think that much lol!",
			"DELETE": delete.__func__,
		}),
	})

	pass


class ApiTags(ApiBase):
	"""
	# `ApiTags`: Tagging API

	- Base Path: `/api/tags`
	- Item Path: `/api/tags/{item}`

	## Methods:

	#### Base Path:

	- `GET`: Returns list of tags that are available (registered).

	#### Item Path:

	- `GET`: Returns list of posts (identifiers) that are registered to item.
	- `PUT`: Tries to create tag (with optional body).
		- `PUT tags/test` with empty body `` or `[]` yields empty tag to be created.
		- `PUT tags/test` with body `[ "postid" ]` or `[ "id_1", "id_2" ]` yields tag created with given contents.
		- Note: In situation tag exists, throws error.
	- `POST`: Adds given list (body) to tag, items that are already added won't be added (no duplicated items). Returns added items.
	- `PATCH`: Deletes items from given tag. Non-existent items won't be removed. Returns list of items that are removed.
	- `DELETE`: Removes tag and its contents altogether, returns the tag thats removed.

	"""

	path = os.path.join(base_path, "tags")
	name = lambda ext="json": (".".join([ ApiTags.path, ext ]))
	load = lambda *args: json.load(S3.obj(ApiTags.name()).get()["Body"])
	save = lambda data: S3.obj(ApiTags.name()).put(
		ACL="public-read", Body=json.dumps(data).encode()
	)
	getk = lambda e: (" ".join(re.findall("[^/]+", e["path"])))

	def __init__(self, *args):
		return #print(__class__, self, args, self.routes)

	def __call__(self, e: dict, m):
		return helper_apicall(e, m, self.routes)

	@staticmethod
	def posts(e, m):
		try:
			key = ApiTags.getk(e)
			try:
				tags: DotDict = ApiTags.load()
			except:
				tags = DotDict()
			return (200, {}, tags[key] if key in tags.keys() else [])
		except Exception as err:
			return (400, {}, { "error": str(err), })
			pass
		return (400, {}, None)

	@staticmethod
	def listall(e, m):
		try:
			return (200, {}, list(ApiTags.load().keys()))
		except Exception as err:
			return (400, {}, { "error": str(err), })
			pass
		return (400, {}, None)

	@staticmethod
	def add(e, m):
		try:
			key = ApiTags.getk(e)
			try:
				tags: DotDict = ApiTags.load()
			except:
				tags = DotDict()
			if key in tags.keys():
				raise Exception("tag alerady exists")
			tags[key] = []
			try:
				tags[key].extend(json.loads(e["body"]))
			except:
				pass
			return (200, {}, ApiTags.save(tags))
		except Exception as err:
			return (400, {}, { "error": str(err), })
			pass
		return (400, {}, None)

	@staticmethod
	def remove(e, m):
		try:
			key = ApiTags.getk(e)
			try:
				tags: DotDict = ApiTags.load()
			except:
				tags = DotDict()
			res = tags.pop(key, None)
			ApiTags.save(tags)
			return (200, {}, { "result": res, })
		except Exception as err:
			return (400, {}, { "error": str(err), })
			pass
		return (400, {}, None)

	@staticmethod
	def bind(e, m):
		try:
			body = json.loads(e["body"])
			if not isinstance(body, list):
				raise Exception("body must be json-list")
			key = ApiTags.getk(e)
			try:
				tags: DotDict = ApiTags.load()
			except:
				tags = DotDict()
			post: list = tags[key]
			res = [
				(item, post.append(item))[0]
				for item in body if item not in post
			]
			ApiTags.save(tags)
			return (200, {}, res)
		except Exception as err:
			return (400, {}, { "error": str(err), })
			pass
		return (400, {}, None)

	@staticmethod
	def unbind(e, m):
		try:
			body = json.loads(e["body"])
			if not isinstance(body, list):
				raise Exception("body must be json-list")
			try:
				tags: DotDict = ApiTags.load()
			except:
				tags = DotDict()
			tags: DotDict = ApiTags.load()
			post: list = tags[key]
			res = [
				(item, post.remove(item))[0]
				for item in body if item in post
			]
			ApiTags.save(tags)
			return (200, {}, res)
		except Exception as err:
			return (400, {}, { "error": str(err), })
			pass
		return (400, {}, None)

	routes = DotDict({
		"^[/]?$": DotDict({
			"GET": listall.__func__,
			"DEF": (200, {}, { "error": "method not supported!", })
		}),
		"^([/]+[^/]+)+$": DotDict({
			"PUT": add.__func__,
			"GET": posts.__func__,
			"POST": bind.__func__,
			"PATCH": unbind.__func__,
			"DELETE": remove.__func__,
		})
	})
	pass

class ApiAttach(ApiBase):
	path = os.path.join(base_path, "attach")
	pass

routes_base = DotDict({
	"^/(api[/]?)?$": "# default route #",
	"^/api/tags([/]+.*)?$": ApiTags(),
	"^/api/posts([/]+.*)?$": ApiPosts(),
	"^/api/tests([/]+.*)?$": ApiBase(),
	#"^/(api[/]?)$": DotDict(message="not implemented (yet)"),
})


def helper_apicall(e, m, r):
	groups = m.groups()
	paths = re.findall("([/]?[^/]*)", (
		groups[0] if groups[0] is not None else "/"
	))
	e["path"] = ("/".join(filter(lambda x: x, paths)))
	return tuple((find(execute, e, r) or dict(
		statusCode = 400,
		headers    = { "content-type": "application/json" },
		body       = { "error": "no route" },
	)).values())

def execute(event, match, function):
	if callable(function):
		(status, headers, body) = function(event, match)
		headers["access-control-allow-origin"] = "*"
		if not isinstance(body, str):
			headers["content-type"] = "application/json"
			body = json.dumps(body, indent='\t')
		return dict(
			statusCode = status,
			headers    = headers,
			body       = body,
		)
	return dict(
		statusCode = 200,
		headers    = { "content-type": "application/json" },
		body       = json.dumps(function, indent='\t')
	)

def find(function, http, routes):
	for k, v in routes.items():
		match = re.match(k, http["path"])
		if not match:
			continue
		if callable(v):
			return function(http, match, v)
		if isinstance(v, dict):
			method = http["httpMethod"]
			if method in list(v.keys()):
				return function(http, match, v[method])
			elif "DEF" in v.keys():
				return function(http, match, v["DEF"])
			pass
		if isinstance(v, str):
			return dict(
				statusCode = 200,
				headers    = { "content-type": "text/plain" },
				body       = v,
			)
		if isinstance(v, list):
			return dict(
				statusCode = 200,
				headers    = { "content-type": "application/json" },
				body       = json.dumps(v, indent='\t')
			)
		if isinstance(v, tuple):
			return dict([
				("statusCode", v[0]),
				("headers",    v[1]),
				("body",       v[2]),
			])
		continue
	return None

def handler(event, context):
	return find(execute, event, routes_base) or dict(statusCode=404)
	return {
		"statusCode": 200,
		"headers": {
			"content-type": "application/json",
		},
		"body": json.dumps({
			"event": event,
			"context": {
				"aws_request_id": context.aws_request_id,
				"client_context": context.client_context,
				"function_name": context.function_name,
				"function_version": context.function_version,
				#"get_remaining_time_in_millis": context.get_remaining_time_in_millis,
				#"identity": context.identity,
				"invoked_function_arn": context.invoked_function_arn,
				#"log": context.log,
				"log_group_name": context.log_group_name,
				"log_stream_name": context.log_stream_name,
				"memory_limit_in_mb": context.memory_limit_in_mb,
			},
		}, indent='\t'),
	}