119 lines
3.7 KiB
Python
119 lines
3.7 KiB
Python
from typing import Optional, List, Union, Literal
|
|
|
|
from pydantic import Field, EmailStr, field_validator, model_validator
|
|
|
|
from chalicelib.utils.TimeUTC import TimeUTC
|
|
from . import schemas
|
|
from .overrides import BaseModel, Enum
|
|
from .transformers_validators import remove_whitespace
|
|
|
|
|
|
class Permissions(str, Enum):
|
|
SESSION_REPLAY = "SESSION_REPLAY"
|
|
DEV_TOOLS = "DEV_TOOLS"
|
|
# errors = "ERRORS"
|
|
METRICS = "METRICS"
|
|
ASSIST_LIVE = "ASSIST_LIVE"
|
|
ASSIST_CALL = "ASSIST_CALL"
|
|
FEATURE_FLAGS = "FEATURE_FLAGS"
|
|
SPOT = "SPOT"
|
|
SPOT_PUBLIC = "SPOT_PUBLIC"
|
|
|
|
|
|
class ServicePermissions(str, Enum):
|
|
SESSION_REPLAY = "SERVICE_SESSION_REPLAY"
|
|
DEV_TOOLS = "SERVICE_DEV_TOOLS"
|
|
ASSIST_LIVE = "SERVICE_ASSIST_LIVE"
|
|
ASSIST_CALL = "SERVICE_ASSIST_CALL"
|
|
READ_NOTES = "SERVICE_READ_NOTES"
|
|
|
|
|
|
class ValidIdentityProviderPermissions(str, Enum):
|
|
SESSION_REPLAY = "SESSION_REPLAY"
|
|
METRICS = "METRICS"
|
|
ASSIST_LIVE = "ASSIST_LIVE"
|
|
ASSIST_CALL = "ASSIST_CALL"
|
|
SPOT_PUBLIC = "SPOT_PUBLIC"
|
|
|
|
|
|
class CurrentContext(schemas.CurrentContext):
|
|
permissions: List[Union[Permissions, ServicePermissions]] = Field(...)
|
|
service_account: bool = Field(default=False)
|
|
|
|
@model_validator(mode="before")
|
|
@classmethod
|
|
def remove_unsupported_perms(cls, values):
|
|
if values.get("permissions") is not None:
|
|
perms = []
|
|
for p in values["permissions"]:
|
|
if Permissions.has_value(p) or ServicePermissions.has_value(p):
|
|
perms.append(p)
|
|
values["permissions"] = perms
|
|
return values
|
|
|
|
|
|
class RolePayloadSchema(BaseModel):
|
|
name: str = Field(..., min_length=1, max_length=40)
|
|
description: Optional[str] = Field(default=None)
|
|
permissions: List[Permissions] = Field(...)
|
|
all_projects: bool = Field(default=True)
|
|
projects: List[int] = Field(default=[])
|
|
_transform_name = field_validator('name', mode="before")(remove_whitespace)
|
|
|
|
|
|
class SignalsSchema(BaseModel):
|
|
timestamp: int = Field(...)
|
|
action: str = Field(...)
|
|
source: str = Field(...)
|
|
category: str = Field(...)
|
|
data: dict = Field(default={})
|
|
|
|
|
|
class CreateMemberSchema(schemas.CreateMemberSchema):
|
|
roleId: Optional[int] = Field(default=None)
|
|
|
|
|
|
class EditMemberSchema(BaseModel):
|
|
name: str = Field(...)
|
|
email: EmailStr = Field(...)
|
|
admin: bool = Field(default=False)
|
|
roleId: int = Field(...)
|
|
|
|
|
|
class TrailSearchPayloadSchema(schemas._PaginatedSchema):
|
|
limit: int = Field(default=200, gt=0)
|
|
startDate: int = Field(default=TimeUTC.now(-7))
|
|
endDate: int = Field(default=TimeUTC.now(1))
|
|
user_id: Optional[int] = Field(default=None)
|
|
query: Optional[str] = Field(default=None)
|
|
action: Optional[str] = Field(default=None)
|
|
order: schemas.SortOrderType = Field(default=schemas.SortOrderType.DESC)
|
|
|
|
@model_validator(mode="before")
|
|
@classmethod
|
|
def transform_order(cls, values):
|
|
if values.get("order") is None:
|
|
values["order"] = schemas.SortOrderType.DESC
|
|
else:
|
|
values["order"] = values["order"].upper()
|
|
return values
|
|
|
|
|
|
class AssistRecordUpdatePayloadSchema(BaseModel):
|
|
name: str = Field(..., min_length=1)
|
|
_transform_name = field_validator('name', mode="before")(remove_whitespace)
|
|
|
|
|
|
class AssistRecordPayloadSchema(AssistRecordUpdatePayloadSchema):
|
|
duration: int = Field(...)
|
|
session_id: int = Field(...)
|
|
|
|
|
|
class AssistRecordSavePayloadSchema(AssistRecordPayloadSchema):
|
|
key: str = Field(...)
|
|
|
|
|
|
class AssistRecordSearchPayloadSchema(schemas._PaginatedSchema, schemas._TimedSchema):
|
|
user_id: Optional[int] = Field(default=None)
|
|
query: Optional[str] = Field(default=None)
|
|
order: Literal["asc", "desc"] = Field(default="desc")
|