39 lines
1.1 KiB
Python
39 lines
1.1 KiB
Python
import bson
|
|
import pydantic
|
|
|
|
from app.storage.mongo import database
|
|
from bson import codec_options
|
|
|
|
|
|
collection = database.get_collection("configs", codec_options=codec_options.CodecOptions(tz_aware=True))
|
|
|
|
|
|
class Config(pydantic.BaseModel):
|
|
name: str
|
|
project: str
|
|
stage: str
|
|
value: dict
|
|
_id: bson.ObjectId|None = None
|
|
|
|
|
|
async def create(config: Config) -> str:
|
|
result = await collection.insert_one(config.model_dump())
|
|
return result.inserted_id
|
|
|
|
|
|
async def update_data(project: str, stage: str, name: str, value: dict) -> bool:
|
|
result = await collection.update_one({'project': project, 'stage': stage, 'name': name}, {'$set': {'value': value}})
|
|
return result.modified_count != 0
|
|
|
|
|
|
async def delete(project: str, stage: str, name: str) -> bool:
|
|
result = await collection.delete_one({'project': project, 'stage': stage, 'name': name})
|
|
return result.deleted_count != 0
|
|
|
|
|
|
async def get(project: str, stage: str) -> list[Config]:
|
|
result = []
|
|
async for item in collection.find({'stage': stage, 'project': project}):
|
|
result.append(Config.model_validate(item))
|
|
return result
|