58 lines
1.4 KiB
Python
58 lines
1.4 KiB
Python
from datetime import datetime
|
|
|
|
from bson import ObjectId
|
|
import motor.motor_asyncio as motor
|
|
from .settings import settings
|
|
|
|
client: motor.AsyncIOMotorClient = motor.AsyncIOMotorClient(settings.mongo_uri)
|
|
|
|
db = client["controlpanel"]
|
|
events: motor.AsyncIOMotorCollection = db["events"]
|
|
|
|
|
|
async def create_event(type: str, **kwargs: any):
|
|
event = {
|
|
"timestamp": datetime.now(),
|
|
"type": type,
|
|
"value": kwargs,
|
|
}
|
|
new_event = await events.insert_one(event)
|
|
created_event = await events.find_one({"_id": new_event.inserted_id})
|
|
return created_event
|
|
|
|
|
|
computers = db["computers"]
|
|
|
|
open_tokens = db["open_tokens"]
|
|
|
|
|
|
async def on_startup():
|
|
await open_tokens.create_index("timestamp", expireAfterSeconds=300)
|
|
if not settings.dev_mode:
|
|
await create_event("server_start")
|
|
|
|
|
|
async def open_token(token: str):
|
|
await open_tokens.insert_one({"timestamp": datetime.now(), "token": token})
|
|
|
|
|
|
async def close_token(token: str):
|
|
deleted = await open_tokens.delete_one({"token": token})
|
|
return deleted.deleted_count
|
|
|
|
|
|
class PyObjectId(ObjectId):
|
|
@classmethod
|
|
def __get_validators__(cls):
|
|
yield cls.validate
|
|
|
|
@classmethod
|
|
def validate(cls, v):
|
|
if not ObjectId.is_valid(v):
|
|
raise ValueError("Invalid objectid")
|
|
return ObjectId(v)
|
|
|
|
@classmethod
|
|
def __modify_schema__(cls, field_schema):
|
|
field_schema.update(type="string")
|