87 lines
2.7 KiB
Python
87 lines
2.7 KiB
Python
from based.db import DBConnector
|
|
from based.columns import (
|
|
ColumnDefinition,
|
|
PrimarySerialColumnDefinition,
|
|
TextColumnDefinition,
|
|
BooleanColumnDefinition,
|
|
TimestampColumnDefinition,
|
|
DoubleColumnDefinition,
|
|
IntegerColumnDefinition,
|
|
)
|
|
from db_addendum import AssetRefColumnDefinition, UserRefColumnDefinition
|
|
import dba
|
|
|
|
|
|
def check_if_admin_access_token(
|
|
connector: DBConnector, access_token: str | None
|
|
) -> bool:
|
|
if access_token is None:
|
|
return False
|
|
|
|
user, group = dba.get_user_by_access_token(connector, access_token)
|
|
if user is None or group is None or group.name != "admin":
|
|
return False
|
|
|
|
return True
|
|
|
|
|
|
def get_column_from_definition(definition: str) -> ColumnDefinition | None:
|
|
match definition.split(":"):
|
|
case [name, "serial", "primary"]:
|
|
return PrimarySerialColumnDefinition(name)
|
|
|
|
case [name, "str", *rest]:
|
|
is_unique = "unique" in rest
|
|
has_default = "default" in rest
|
|
td = TextColumnDefinition(name, unique=is_unique)
|
|
td.has_default = has_default
|
|
return td
|
|
|
|
case [name, "bool", *rest]:
|
|
is_unique = "unique" in rest
|
|
has_default = "default" in rest
|
|
td = BooleanColumnDefinition(name, unique=is_unique)
|
|
td.has_default = has_default
|
|
return td
|
|
|
|
case [name, "datetime", *rest]:
|
|
is_unique = "unique" in rest
|
|
has_default = "default" in rest
|
|
td = TimestampColumnDefinition(name, unique=is_unique)
|
|
td.has_default = has_default
|
|
return td
|
|
|
|
case [name, "float", *rest]:
|
|
is_unique = "unique" in rest
|
|
has_default = "default" in rest
|
|
td = DoubleColumnDefinition(name, unique=is_unique)
|
|
td.has_default = has_default
|
|
return td
|
|
|
|
case [name, "int", *rest]:
|
|
is_unique = "unique" in rest
|
|
has_default = "default" in rest
|
|
td = IntegerColumnDefinition(name, unique=is_unique)
|
|
td.has_default = has_default
|
|
return td
|
|
|
|
case [name, "int-user", *rest]:
|
|
return UserRefColumnDefinition(name)
|
|
|
|
case [name, "int-asset", *rest]:
|
|
return AssetRefColumnDefinition(name)
|
|
|
|
return None
|
|
|
|
|
|
def parse_columns_from_definition(definition: str) -> list[ColumnDefinition]:
|
|
columns = []
|
|
for column_definition in definition.split(","):
|
|
if column_definition == "":
|
|
continue
|
|
column = get_column_from_definition(column_definition)
|
|
if column is None:
|
|
raise ValueError(f"Invalid column definition: {column_definition}")
|
|
columns.append(column)
|
|
|
|
return columns
|