Doorboy proxy exports door access information from MongoDB to door Raspberry Pi controllers
You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.

74 lines
2.3 KiB

from sanic import Sanic
from sanic.response import text, json
from motor.motor_asyncio import AsyncIOMotorClient
import pymongo
import os
app = Sanic(__name__)
MONGO_URI = os.getenv("MONGO_URI",
assert len(DOORBOY_SECRET) >= 10
async def setup_db(app, loop):
# TODO: find cleaner way to do this, for more see
app.ctx.db = AsyncIOMotorClient(MONGO_URI).get_default_database()
async def view_doorboy_uids(request):
if request.headers.get("KEY") != DOORBOY_SECRET:
return text("how about no")
allowed_names = []
async for obj in app.ctx.db.member.find({"enabled": True}):
allowed_uids = []
flt = {
"token.uid_hash": {"$exists": True},
"inventory.owner": {"$exists": True},
"token.enabled": {"$exists": True}
prj = {
"inventory.owner": True,
"token.uid_hash": True
async for obj in app.ctx.db.inventory.find(flt, prj):
if obj["inventory"].pop("owner").get("foreign_id") in allowed_names:
del obj["_id"]
del obj["inventory"]
return json({"allowed_uids": allowed_uids})
@app.route("/longpoll", stream=True)
async def view_longpoll(request):
response = await request.respond(content_type="text/event-stream")
if request.headers.get("KEY") != DOORBOY_SECRET:
return text("Invalid token")
await response.send("data: response-generator-started\n\n")
pipeline = [
"$match": {
"operationType": "insert",
async with as stream:
await response.send("data: watch-stream-opened\n\n")
async for event in stream:
if event["fullDocument"].get("type") == "open-door":
await response.send("data: %s\n\n" %
except pymongo.errors.PyMongoError:
if __name__ == "__main__":, host="", port=5000)