61 lines
1.5 KiB
Python
61 lines
1.5 KiB
Python
"""Route dependencies."""
|
|
|
|
from datetime import datetime as dt
|
|
from datetime import timedelta
|
|
|
|
import owncloud
|
|
from fastapi import Depends, Request
|
|
from humanize import naturaldelta
|
|
|
|
|
|
async def fresh_token(request: Request):
|
|
"""Ensure fresh credentials for speaking to Keycloak."""
|
|
from members_lumbung_space.keycloak import init_keycloak
|
|
|
|
request.app.state.keycloak = init_keycloak()
|
|
|
|
|
|
async def logged_in(request: Request):
|
|
"""Ensure the user is logged in."""
|
|
from members_lumbung_space.exceptions import RequiresLoginException
|
|
|
|
user = request.session.get("user")
|
|
|
|
if not user:
|
|
raise RequiresLoginException
|
|
|
|
return user
|
|
|
|
|
|
async def get_user(request: Request):
|
|
"""Retrieve the user object."""
|
|
return request.session.get("user")
|
|
|
|
|
|
async def get_invites(request: Request, user=Depends(get_user)):
|
|
"""Retrieve all invites from storage."""
|
|
from members_lumbung_space.config import INVITE_TIME_LIMIT
|
|
|
|
all_invites = {}
|
|
|
|
for username in await request.app.state.redis.keys("*"):
|
|
if username == "resource_map":
|
|
continue
|
|
|
|
invites = await request.app.state.redis.get(username)
|
|
|
|
for invite in invites:
|
|
invite["validity"] = naturaldelta(
|
|
dt.fromisoformat(invite["time"])
|
|
+ timedelta(days=INVITE_TIME_LIMIT)
|
|
)
|
|
|
|
all_invites[username] = invites
|
|
|
|
return all_invites
|
|
|
|
|
|
async def get_resource_map(request: Request):
|
|
"""Retrieve the resource map listing."""
|
|
return await request.app.state.redis.get("resource_map")
|