Module licenseware.app_builder.uploads_namespace.status_namespace

Expand source code
from flask import request
from flask_restx import Namespace, Resource
from licenseware.decorators.auth_decorators import authorization_check
from licenseware.decorators import failsafe

# from licenseware.tenants.processing_status import get_uploader_status #! OUTDATED
from licenseware.notifications.uploader_status import get_uploader_status


from licenseware.uploader_builder import UploaderBuilder
from typing import List


def create_uploader_resource(uploader: UploaderBuilder):
    class UploaderStatus(Resource):
        @failsafe(fail_code=500)
        @authorization_check
        def get(self):
            return get_uploader_status(
                request.headers.get("Tenantid"), uploader.uploader_id
            )
            # return get_uploader_status(request.headers.get("Tenantid"), uploader.uploader_id)

    return UploaderStatus


def get_status_namespace(ns: Namespace, uploaders: List[UploaderBuilder]):

    for uploader in uploaders:

        UR = create_uploader_resource(uploader)

        @ns.doc(
            description="Get processing status of files uploaded",
            responses={
                200: "idle or running",
                400: "Tenantid not provided",
                403: "Missing `Tenant` or `Authorization` information",
                500: "Something went wrong while handling the request",
            },
        )
        class TempUploaderResource(UR):
            ...

        UploaderResource = type(
            uploader.uploader_id.replace("_", "").capitalize() + "status",
            (TempUploaderResource,),
            {},
        )

        ns.add_resource(UploaderResource, uploader.status_check_path)

    return ns

Functions

def create_uploader_resource(uploader: UploaderBuilder)
Expand source code
def create_uploader_resource(uploader: UploaderBuilder):
    class UploaderStatus(Resource):
        @failsafe(fail_code=500)
        @authorization_check
        def get(self):
            return get_uploader_status(
                request.headers.get("Tenantid"), uploader.uploader_id
            )
            # return get_uploader_status(request.headers.get("Tenantid"), uploader.uploader_id)

    return UploaderStatus
def get_status_namespace(ns: flask_restx.namespace.Namespace, uploaders: List[UploaderBuilder])
Expand source code
def get_status_namespace(ns: Namespace, uploaders: List[UploaderBuilder]):

    for uploader in uploaders:

        UR = create_uploader_resource(uploader)

        @ns.doc(
            description="Get processing status of files uploaded",
            responses={
                200: "idle or running",
                400: "Tenantid not provided",
                403: "Missing `Tenant` or `Authorization` information",
                500: "Something went wrong while handling the request",
            },
        )
        class TempUploaderResource(UR):
            ...

        UploaderResource = type(
            uploader.uploader_id.replace("_", "").capitalize() + "status",
            (TempUploaderResource,),
            {},
        )

        ns.add_resource(UploaderResource, uploader.status_check_path)

    return ns