127 lines
4.7 KiB
Python
127 lines
4.7 KiB
Python
from based.db import DBConnector
|
|
from based.columns import (
|
|
ColumnDefinition,
|
|
make_column_unique,
|
|
PrimarySerialColumnDefinition,
|
|
PrimaryUUIDColumnDefinition,
|
|
TextColumnDefinition,
|
|
BigintColumnDefinition,
|
|
BooleanColumnDefinition,
|
|
DateColumnDefinition,
|
|
TimestampColumnDefinition,
|
|
DoubleColumnDefinition,
|
|
IntegerColumnDefinition,
|
|
UUIDColumnDefinition,
|
|
)
|
|
from dba import get_user_by_access_token
|
|
|
|
|
|
def check_if_admin_access_token(
|
|
connector: DBConnector, access_token: str | None
|
|
) -> bool:
|
|
if access_token is None:
|
|
return False
|
|
|
|
user, group = 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, "uuid", "primary"]:
|
|
return PrimaryUUIDColumnDefinition(name)
|
|
|
|
case [name, "str"]:
|
|
return TextColumnDefinition(name)
|
|
case [name, "str", "unique"]:
|
|
return make_column_unique(TextColumnDefinition(name))
|
|
case [name, "str", "default", default]:
|
|
return TextColumnDefinition(name, default=default)
|
|
case [name, "str", "default", default, "unique"]:
|
|
return make_column_unique(TextColumnDefinition(name, default=default))
|
|
|
|
case [name, "bigint"]:
|
|
return BigintColumnDefinition(name)
|
|
case [name, "bigint", "unique"]:
|
|
return make_column_unique(BigintColumnDefinition(name))
|
|
case [name, "bigint", "default", default]:
|
|
return BigintColumnDefinition(name, default=int(default))
|
|
case [name, "bigint", "default", default, "unique"]:
|
|
return make_column_unique(
|
|
BigintColumnDefinition(name, default=int(default))
|
|
)
|
|
|
|
case [name, "bool"]:
|
|
return BooleanColumnDefinition(name)
|
|
case [name, "bool", "unique"]:
|
|
return make_column_unique(BooleanColumnDefinition(name))
|
|
case [name, "bool", "default", default]:
|
|
return BooleanColumnDefinition(name, default=bool(default))
|
|
case [name, "bool", "default", default, "unique"]:
|
|
return make_column_unique(
|
|
BooleanColumnDefinition(name, default=bool(default))
|
|
)
|
|
|
|
case [name, "date"]:
|
|
return DateColumnDefinition(name)
|
|
case [name, "date", "unique"]:
|
|
return make_column_unique(DateColumnDefinition(name))
|
|
# TODO: Add default value for date
|
|
|
|
case [name, "datetime"]:
|
|
return TimestampColumnDefinition(name)
|
|
case [name, "datetime", "unique"]:
|
|
return make_column_unique(TimestampColumnDefinition(name))
|
|
# TODO: Add default value for timestamp
|
|
|
|
case [name, "float"]:
|
|
return DoubleColumnDefinition(name)
|
|
case [name, "float", "unique"]:
|
|
return make_column_unique(DoubleColumnDefinition(name))
|
|
case [name, "float", "default", default]:
|
|
return DoubleColumnDefinition(name, default=float(default))
|
|
case [name, "float", "default", default, "unique"]:
|
|
return make_column_unique(
|
|
DoubleColumnDefinition(name, default=float(default))
|
|
)
|
|
|
|
case [name, "int"]:
|
|
return IntegerColumnDefinition(name)
|
|
case [name, "int", "unique"]:
|
|
return make_column_unique(IntegerColumnDefinition(name))
|
|
case [name, "int", "default", default]:
|
|
return IntegerColumnDefinition(name, default=int(default))
|
|
case [name, "int", "default", default, "unique"]:
|
|
return make_column_unique(
|
|
IntegerColumnDefinition(name, default=int(default))
|
|
)
|
|
|
|
case [name, "uuid"]:
|
|
return UUIDColumnDefinition(name)
|
|
case [name, "uuid", "unique"]:
|
|
return make_column_unique(UUIDColumnDefinition(name))
|
|
case [name, "uuid", "default", default]:
|
|
return UUIDColumnDefinition(name, default=default)
|
|
case [name, "uuid", "default", default, "unique"]:
|
|
return make_column_unique(UUIDColumnDefinition(name, default=default))
|
|
|
|
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
|