arbor_imago.services package

Submodules

arbor_imago.services.api_key module

class arbor_imago.services.api_key.ApiKey(*args, **kwargs)[source]

Bases: Service[ApiKey, str, ApiKeyAdminCreate, ApiKeyAdminUpdate, Literal[‘issued’, ‘expiry’, ‘name’]], SimpleIdModelService[ApiKey, str], JwtIO[ApiKey, str], Table[ApiKey], JwtAndSimpleIdTable[ApiKey, str]

auth_type: ClassVar[auth_credential_schema.Type] = 'api_key'

async classmethod get_scope_ids(session, inst)[source]

async classmethod is_available(session: AsyncSession, api_key_available_admin: ApiKeyAdminAvailable) → bool[source]

classmethod model_inst_from_create_model(create_model)[source]

arbor_imago.services.api_key_scope module

class arbor_imago.services.api_key_scope.ApiKeyScope(*args, **kwargs)[source]

Bases: Service[ApiKeyScope, ApiKeyScopeId, ApiKeyScopeAdminCreate, ApiKeyScopeAdminUpdate, str]

classmethod model_id(inst: ApiKeyScope)[source]

arbor_imago.services.auth_credential module

class arbor_imago.services.auth_credential.HasAuthType(*args, **kwargs)[source]

Bases: Protocol[TAuthCredential]

auth_type: ClassVar[Type]

class arbor_imago.services.auth_credential.HasModelInstFromJwtPayload(*args, **kwargs)[source]

Bases: Protocol[TAuthCredentialJwt_co, TSub]

classmethod model_inst_from_jwt_payload(payload: JwtPayload[TSub]) → TAuthCredentialJwt_co[source]

class arbor_imago.services.auth_credential.HasModelSub(*args, **kwargs)[source]: Bases: Protocol[TAuthCredentialJwt_contra, TSub_co]

class arbor_imago.services.auth_credential.JwtAndSimpleIdTable(*args, **kwargs)[source]: Bases: Generic[TAuthCredentialJwtAndTable, TSimpleId], HasModelSub[TAuthCredentialJwtAndTable, TSimpleId], HasModelId[TAuthCredentialJwtAndTable, TSimpleId]

class arbor_imago.services.auth_credential.JwtIO(*args, **kwargs)[source]

Bases: Generic[TAuthCredentialJwt, TSub], HasAuthType[TAuthCredentialJwt], HasModelSub[TAuthCredentialJwt, TSub]

classmethod to_jwt_payload(inst: TAuthCredentialJwt) → JwtPayload[TSub][source]

classmethod validate_jwt_claims(payload: JwtPayload[TSub])[source]

class arbor_imago.services.auth_credential.JwtNotTable(*args, **kwargs)[source]: Bases: Generic[TAuthCredentialJwtAndNotTable, TSub, TCreateModel], HasModelInstFromJwtPayload[TAuthCredentialJwtAndNotTable, TSub], HasModelInstFromCreateModel[TAuthCredentialJwtAndNotTable, TCreateModel], HasModel[TAuthCredentialJwtAndNotTable], HasModelSub[TAuthCredentialJwtAndNotTable, TSub]

exception arbor_imago.services.auth_credential.MissingRequiredClaimsError(claims: set[str])[source]: Bases: Exception

class arbor_imago.services.auth_credential.Table(*args, **kwargs)[source]

Bases: Generic[TAuthCredentialTable], HasAuthType[TAuthCredentialTable]

async classmethod get_scope_ids(session: AsyncSession, inst: TAuthCredentialTable) → list[int][source]

arbor_imago.services.auth_credential.lifespan_to_expiry(lifespan: timedelta) → Annotated[datetime, 'The datetime at which the auth credential will expire'][source]

arbor_imago.services.base module

exception arbor_imago.services.base.AlreadyExistsError(model: Type[User | UserAccessToken | OTP | ApiKey | Gallery | File | ImageVersion | ApiKeyScope | GalleryPermission | ImageFileMetadata | SignUp], id: str | GalleryPermissionId | ApiKeyScopeId)[source]: Bases: ServiceError

class arbor_imago.services.base.CRUDParamsBase[source]

Bases: TypedDict

admin: bool

authorized_user_id: str | None

session: AsyncSession

class arbor_imago.services.base.CheckAuthorizationExistingParams[source]

Bases: Generic[TModel_contra, TId], CRUDParamsBase, WithId[TId], WithModelInst[TModel_contra]

admin: bool

authorized_user_id: str | None

id: TId

model_inst: TModel_contra

operation: Literal['read', 'update', 'delete']

session: AsyncSession

class arbor_imago.services.base.CheckAuthorizationNewParams[source]

Bases: Generic[TCreateModel_contra], CreateParams[TCreateModel_contra]

admin: bool

authorized_user_id: str | None

create_model: TCreateModel_contra

session: AsyncSession

class arbor_imago.services.base.CheckAuthorizationReadManyParams[source]

Bases: Generic[TModel, TOrderBy_co], ReadManyParams[TModel, TOrderBy_co]

admin: bool

authorized_user_id: str | None

order_bys: NotRequired[list[OrderBy[TypeVar]]]

pagination: Pagination

query: NotRequired[SelectOfScalar | None]

session: AsyncSession

class arbor_imago.services.base.CheckValidationDeleteParams[source]

Bases: Generic[TId], DeleteParams[TId]

admin: bool

authorized_user_id: str | None

id: TId

session: AsyncSession

class arbor_imago.services.base.CheckValidationPatchParams[source]

Bases: Generic[TModel, TId, TUpdateModel_contra], UpdateParams[TId, TUpdateModel_contra], WithModelInst[TModel]

admin: bool

authorized_user_id: str | None

id: TId

model_inst: TModel_contra

session: AsyncSession

update_model: TUpdateModel_contra

class arbor_imago.services.base.CheckValidationPostParams[source]

Bases: Generic[TCreateModel_contra], CreateParams[TCreateModel_contra]

admin: bool

authorized_user_id: str | None

create_model: TCreateModel_contra

session: AsyncSession

class arbor_imago.services.base.CreateParams[source]

Bases: Generic[TCreateModel_contra], CRUDParamsBase

admin: bool

authorized_user_id: str | None

create_model: TCreateModel_contra

session: AsyncSession

class arbor_imago.services.base.DeleteParams[source]

Bases: Generic[TId], CRUDParamsBase, WithId[TId]

admin: bool

authorized_user_id: str | None

id: TId

session: AsyncSession

class arbor_imago.services.base.HasBuildSelectById(*args, **kwargs)[source]: Bases: Protocol[TModel, TId_contra]

class arbor_imago.services.base.HasModel(*args, **kwargs)[source]: Bases: Protocol[TModel_co]

class arbor_imago.services.base.HasModelId(*args, **kwargs)[source]

Bases: Protocol[TModel_contra, TId_co]

classmethod model_id(inst: TModel_contra) → TId_co[source]

class arbor_imago.services.base.HasModelInstFromCreateModel(*args, **kwargs)[source]

Bases: Protocol[TModel_co, TCreateModel_contra]

classmethod model_inst_from_create_model(create_model: TCreateModel_contra) → TModel_co[source]

exception arbor_imago.services.base.NotAvailableError(error_message: str)[source]: Bases: ServiceError

Bases: ValueError, ServiceError

static not_found_message(model: Type[User | UserAccessToken | OTP | ApiKey | Gallery | File | ImageVersion | ApiKeyScope | GalleryPermission | ImageFileMetadata | SignUp], id: str | GalleryPermissionId | ApiKeyScopeId) → str[source]

class arbor_imago.services.base.ReadManyBase[source]

Bases: Generic[TModel, TOrderBy_co], TypedDict

order_bys: NotRequired[list[OrderBy[TypeVar]]]

pagination: Pagination

query: NotRequired[SelectOfScalar | None]

class arbor_imago.services.base.ReadManyParams[source]

Bases: Generic[TModel, TOrderBy_co], CRUDParamsBase, ReadManyBase[TModel, TOrderBy_co]

admin: bool

authorized_user_id: str | None

order_bys: NotRequired[list[OrderBy[TypeVar]]]

pagination: Pagination

query: NotRequired[SelectOfScalar | None]

session: AsyncSession

class arbor_imago.services.base.ReadParams[source]

Bases: Generic[TId], CRUDParamsBase, WithId[TId]

admin: bool

authorized_user_id: str | None

id: TId

session: AsyncSession

class arbor_imago.services.base.Service(*args, **kwargs)[source]

Bases: Generic[TModel, TId, TCreateModel, TUpdateModel, TOrderBy_co], HasModel[TModel], HasModelInstFromCreateModel[TModel, TCreateModel], HasModelId[TModel, TId], HasBuildSelectById[TModel, TId]

classmethod build_order_by(query: SelectOfScalar, order_by: list[OrderBy[TypeVar]])[source]

async classmethod create(params: CreateParams[TCreateModel]) → TModel[source]: Used in conjunction with API endpoints, raises exceptions while trying to create a new instance of the model

async classmethod delete(params: DeleteParams[TId]) → None[source]: Used in conjunction with API endpoints, raises exceptions while trying to delete an instance of the model by ID

async classmethod fetch_by_id(session: AsyncSession, id: TId) → TModel | None[source]

async classmethod fetch_by_id_with_exception(session: AsyncSession, id: TId) → TModel[source]

async classmethod fetch_many(session: AsyncSession, pagination: Pagination, order_bys: list[OrderBy[TypeVar]] = [], query: SelectOfScalar | None = None) → Sequence[TModel][source]

async classmethod fetch_one(session: AsyncSession, query: SelectOfScalar) → TModel | None[source]

classmethod model_inst_from_create_model(create_model: TCreateModel) → TModel[source]

async classmethod read(params: ReadParams[TId]) → TModel[source]: Used in conjunction with API endpoints, raises exceptions while trying to get an instance of the model by ID

async classmethod read_many(params: ReadManyParams[TModel, TOrderBy_co]) → Sequence[TModel][source]: Used in conjunction with API endpoints, raises exceptions while trying to get a list of instances of the model

async classmethod update(params: UpdateParams[TId, TUpdateModel]) → TModel[source]: Used in conjunction with API endpoints, raises exceptions while trying to update an instance of the model by ID

exception arbor_imago.services.base.ServiceError(error_message: str)[source]

Bases: Exception

error_message: str

class arbor_imago.services.base.SimpleIdModelService(*args, **kwargs)[source]

Bases: Generic[TSimpleModel, TSimpleId], HasModel[TSimpleModel], HasModelId[TSimpleModel, TSimpleId], HasBuildSelectById[TSimpleModel, TSimpleId]

classmethod model_id(inst: TSimpleModel) → TSimpleId[source]

exception arbor_imago.services.base.UnauthorizedError(error_message: str)[source]: Bases: ServiceError

class arbor_imago.services.base.UpdateParams[source]

Bases: Generic[TId, TUpdateModel_contra], CRUDParamsBase, WithId[TId]

admin: bool

authorized_user_id: str | None

id: TId

session: AsyncSession

update_model: TUpdateModel_contra

class arbor_imago.services.base.WithId[source]

Bases: Generic[TId], TypedDict

id: TId

class arbor_imago.services.base.WithModelInst[source]

Bases: Generic[TModel_contra], TypedDict

model_inst: TModel_contra

arbor_imago.services.file module

class arbor_imago.services.file.File(*args, **kwargs)[source]

Bases: Service[File, str, FileAdminCreate, FileAdminUpdate, str], SimpleIdModelService[File, str]

classmethod model_name(inst: File) → str[source]

arbor_imago.services.gallery module

class arbor_imago.services.gallery.Gallery(*args, **kwargs)[source]

Bases: Service[Gallery, str, GalleryAdminCreate, GalleryAdminUpdate, str], SimpleIdModelService[Gallery, str]

classmethod get_date_and_name_from_folder_name(folder_name: str) → GalleryDateAndName[source]

async classmethod get_dir(session: AsyncSession, gallery: Gallery, root: Path) → Path[source]

async classmethod get_parents(session: AsyncSession, gallery: Gallery) → list[Gallery][source]

async classmethod get_root_gallery(session: AsyncSession, user_id: str) → Gallery | None[source]

async classmethod is_available(session: AsyncSession, gallery_available_admin: GalleryAdminAvailable) → bool[source]

classmethod model_folder_name(inst: Gallery) → str[source]

classmethod model_inst_from_create_model(create_model)[source]

arbor_imago.services.gallery_permission module

class arbor_imago.services.gallery_permission.GalleryPermission(*args, **kwargs)[source]

Bases: Service[GalleryPermission, GalleryPermissionId, GalleryPermissionAdminCreate, GalleryPermissionAdminUpdate, str]

classmethod model_id(inst)[source]

arbor_imago.services.image_file_metadata module

class arbor_imago.services.image_file_metadata.ImageFileMetadata(*args, **kwargs)[source]

Bases: Service[ImageFileMetadata, str, ImageFileMetadataAdminCreate, ImageFileMetadataAdminUpdate, str]

SUFFIXES: ClassVar[set[str]] = {'.bmp', '.gif', '.jpeg', '.jpg', '.png', '.tiff', '.webp'}

classmethod model_id(inst)[source]

classmethod parse_file_stem(file_stem: str) → tuple[Annotated[str, StringConstraints(strip_whitespace=None, to_upper=None, to_lower=None, strict=None, min_length=1, max_length=240, pattern=re.compile('^(?!.*_).+$'))], Annotated[str, StringConstraints(strip_whitespace=None, to_upper=None, to_lower=None, strict=None, min_length=None, max_length=None, pattern=re.compile('^(?!\\d{2}$).+$'))] | None, int | None][source]

arbor_imago.services.image_version module

class arbor_imago.services.image_version.ImageVersion(*args, **kwargs)[source]: Bases: Service[ImageVersion, str, FileAdminCreate, FileAdminUpdate, str], SimpleIdModelService[ImageVersion, str]

arbor_imago.services.otp module

class arbor_imago.services.otp.OTP(*args, **kwargs)[source]

Bases: Service[OTP, str, OTPAdminCreate, OTPAdminUpdate, str], SimpleIdModelService[OTP, str], Table[OTP]

auth_type: ClassVar[auth_credential_schema.Type] = 'otp'

classmethod generate_code() → Annotated[str, StringConstraints(strip_whitespace=None, to_upper=None, to_lower=None, strict=None, min_length=6, max_length=6, pattern=re.compile('^\\d{6}$'))][source]

classmethod hash_code(code: Annotated[str, StringConstraints(strip_whitespace=None, to_upper=None, to_lower=None, strict=None, min_length=6, max_length=6, pattern=re.compile('^\\d{6}$'))]) → str[source]

classmethod model_inst_from_create_model(create_model)[source]

classmethod verify_code(code: Annotated[str, StringConstraints(strip_whitespace=None, to_upper=None, to_lower=None, strict=None, min_length=6, max_length=6, pattern=re.compile('^\\d{6}$'))], hashed_code: str) → bool[source]

arbor_imago.services.sign_up module

class arbor_imago.services.sign_up.SignUp(*args, **kwargs)[source]

Bases: JwtIO[SignUp, Annotated[EmailStr, StringConstraints(strip_whitespace=None, to_upper=None, to_lower=None, strict=None, min_length=1, max_length=254, pattern=None)]], JwtNotTable[SignUp, Annotated[EmailStr, StringConstraints(strip_whitespace=None, to_upper=None, to_lower=None, strict=None, min_length=1, max_length=254, pattern=None)], SignUpAdminCreate]

auth_type: ClassVar[auth_credential_schema.Type] = 'sign_up'

classmethod model_inst_from_create_model(create_model)[source]: Create a new instance of the model from the create model (TCreateModel), don’t overwrite this method

classmethod model_inst_from_jwt_payload(payload)[source]

arbor_imago.services.user module

class arbor_imago.services.user.User(*args, **kwargs)[source]

Bases: Service[User, str, UserAdminCreate, UserAdminUpdate, str], SimpleIdModelService[User, str]

DEFAULT_ROLE_ID = 1

async classmethod authenticate(session: AsyncSession, username_or_email: Annotated[EmailStr, StringConstraints(strip_whitespace=None, to_upper=None, to_lower=None, strict=None, min_length=1, max_length=254, pattern=None)] | Annotated[str, StringConstraints(strip_whitespace=None, to_upper=None, to_lower=True, strict=None, min_length=3, max_length=20, pattern=re.compile('^[a-zA-Z0-9_.-]+$'))], password: Annotated[str, StringConstraints(strip_whitespace=None, to_upper=None, to_lower=None, strict=None, min_length=1, max_length=64, pattern=None)]) → User | None[source]

async classmethod fetch_by_email(session: AsyncSession, email: Annotated[EmailStr, StringConstraints(strip_whitespace=None, to_upper=None, to_lower=None, strict=None, min_length=1, max_length=254, pattern=None)]) → User | None[source]

async classmethod fetch_by_email_or_username(session: AsyncSession, username_or_email: Annotated[EmailStr, StringConstraints(strip_whitespace=None, to_upper=None, to_lower=None, strict=None, min_length=1, max_length=254, pattern=None)] | Annotated[str, StringConstraints(strip_whitespace=None, to_upper=None, to_lower=True, strict=None, min_length=3, max_length=20, pattern=re.compile('^[a-zA-Z0-9_.-]+$'))]) → User | None[source]

async classmethod fetch_by_username(session: AsyncSession, username: Annotated[str, StringConstraints(strip_whitespace=None, to_upper=None, to_lower=True, strict=None, min_length=3, max_length=20, pattern=re.compile('^[a-zA-Z0-9_.-]+$'))]) → User | None[source]

classmethod get_inst_dir(inst: User, root: Path) → Path[source]

classmethod hash_password(password: Annotated[str, StringConstraints(strip_whitespace=None, to_upper=None, to_lower=None, strict=None, min_length=1, max_length=64, pattern=None)]) → str[source]

async classmethod is_email_available(session: AsyncSession, email: Annotated[EmailStr, StringConstraints(strip_whitespace=None, to_upper=None, to_lower=None, strict=None, min_length=1, max_length=254, pattern=None)]) → bool[source]

classmethod is_inst_public(inst: User) → bool[source]

async classmethod is_username_available(session: AsyncSession, username: Annotated[str, StringConstraints(strip_whitespace=None, to_upper=None, to_lower=True, strict=None, min_length=3, max_length=20, pattern=re.compile('^[a-zA-Z0-9_.-]+$'))]) → bool[source]

classmethod model_inst_from_create_model(create_model)[source]

arbor_imago.services.user_access_token module

class arbor_imago.services.user_access_token.UserAccessToken(*args, **kwargs)[source]

Bases: Service[UserAccessToken, str, UserAccessTokenAdminCreate, UserAccessTokenAdminUpdate, str], SimpleIdModelService[UserAccessToken, str], JwtIO[UserAccessToken, str], Table[UserAccessToken], JwtAndSimpleIdTable[UserAccessToken, str]

auth_type: ClassVar[auth_credential_schema.Type] = 'access_token'

async classmethod get_scope_ids(session, inst)[source]

classmethod model_inst_from_create_model(create_model)[source]

Module contents

class arbor_imago.services.AuthCredentialTypeToService[source]

Bases: TypedDict

access_token: Type[UserAccessToken]

api_key: Type[ApiKey]

otp: Type[OTP]

sign_up: Type[SignUp]