Module `licenseware.app_builder.uploads_namespace.filestream_validation_namespace`

Here we have the uploader route builder section. Each time we have a list of entities we need to create a new route or a similar route, like we have below. We need to separate the resource from the restx namespace, otherwise the resource will be overwrited.

Expand source code

"""
Here we have the uploader route builder section.
Each time we have a list of entities we need to create a new route or a similar route, like we have below.
We need to separate the resource from the restx namespace, otherwise the resource will be overwrited.

"""


from statistics import mode
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 import clear_tenant_data
from werkzeug.datastructures import FileStorage
from marshmallow import Schema, fields
from licenseware.common import marshmallow_to_restx_model
from licenseware.utils.common import trigger_broker_funcs

from licenseware.uploader_builder import UploaderBuilder
from typing import List


class FileUploadDetailedValidationSchema(Schema):
    status = fields.Str()
    message = fields.Str()
    filename = fields.Str()
    filepath = fields.Str()


class FileUploadValidationSchema(Schema):
    tenant_id = fields.Str()
    status = fields.Str()
    message = fields.Str()
    validation = fields.List(fields.Nested(FileUploadDetailedValidationSchema))


class FileUploadEventDataSchema(Schema):
    tenant_id = fields.Str()
    uploader_id = fields.Str()
    event_id = fields.Str()
    filepaths = fields.List(fields.Str)
    flask_request = fields.List(fields.Raw)
    validation_response = fields.Nested(FileUploadValidationSchema)


class FileUploadRespSchema(Schema):
    status = fields.Str()
    message = fields.Str()
    #event_data = fields.List(fields.Nested(FileUploadEventDataSchema))
    event_id = fields.Str()


def create_uploader_resource(uploader: UploaderBuilder):
    class FileStreamValidate(Resource):
        @failsafe(fail_code=500)
        @authorization_check
        def post(self):

            clear_data = request.args.get("clear_data", "false")
            if "true" in clear_data.lower():
                clear_tenant_data(
                    request.headers.get("Tenantid"), uploader.collections_list
                )

            upload_response = uploader.upload_files(request)
            if uploader.broker_funcs:
                trigger_broker_funcs(request, uploader.broker_funcs, upload_response=upload_response[0])

            return upload_response


    return FileStreamValidate


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

    file_validation_resp_model = marshmallow_to_restx_model(ns, FileUploadRespSchema)

    file_upload_parser = ns.parser()
    file_upload_parser.add_argument(
        "files[]",
        location="files",
        type=FileStorage,
        required=True,
        # action="append", # Uploading multiple files doesn't work on swagger
        help="Upload files for processing",
    )

    default_params={
        "clear_data": "Boolean parameter, warning, will clear existing data",
        "event_id": "The uuid4 string received on filenames validation",
    }

    for uploader in uploaders:

        if uploader.worker is None:
            continue

        UR = create_uploader_resource(uploader)

        @ns.doc(
            description="Upload files received on `files[]` for processing",
            params=default_params,
            responses={
                400: "File list is empty or files are not on 'files[]' key",
                402: "Quota exceeded",
                403: "Missing `Tenant` or `Authorization` information",
                500: "Something went wrong while handling the request",
            },
        )
        @ns.expect(file_upload_parser)
        @ns.response(
            code=200, description="Upload response", model=file_validation_resp_model
        )
        class TempUploaderResource(UR):
            ...

        # Adding extra swagger query parameters if provided
        if uploader.query_params_on_upload:

            if isinstance(uploader.query_params_on_upload, list):
                params_dict = {}
                for param in uploader.query_params_on_upload:
                    params_dict[param["id"]] = {
                        "description": param["description"],
                        "allowed_values": param["allowed_values"],
                    }
            else:
                params_dict = {}
                for (
                    param_name,
                    param_description,
                ) in uploader.query_params_on_upload.items():
                    params_dict[param_name] = {"description": param_description}

            TempUploaderResource.__apidoc__.update({"params": {**default_params, **params_dict}})

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

        ns.add_resource(UploaderResource, uploader.upload_path)

    return ns

Functions

def create_uploader_resource(uploader: UploaderBuilder)

Expand source code

def create_uploader_resource(uploader: UploaderBuilder):
    class FileStreamValidate(Resource):
        @failsafe(fail_code=500)
        @authorization_check
        def post(self):

            clear_data = request.args.get("clear_data", "false")
            if "true" in clear_data.lower():
                clear_tenant_data(
                    request.headers.get("Tenantid"), uploader.collections_list
                )

            upload_response = uploader.upload_files(request)
            if uploader.broker_funcs:
                trigger_broker_funcs(request, uploader.broker_funcs, upload_response=upload_response[0])

            return upload_response


    return FileStreamValidate

def get_filestream_validation_namespace(ns: flask_restx.namespace.Namespace, uploaders: List[UploaderBuilder])

Expand source code

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

    file_validation_resp_model = marshmallow_to_restx_model(ns, FileUploadRespSchema)

    file_upload_parser = ns.parser()
    file_upload_parser.add_argument(
        "files[]",
        location="files",
        type=FileStorage,
        required=True,
        # action="append", # Uploading multiple files doesn't work on swagger
        help="Upload files for processing",
    )

    default_params={
        "clear_data": "Boolean parameter, warning, will clear existing data",
        "event_id": "The uuid4 string received on filenames validation",
    }

    for uploader in uploaders:

        if uploader.worker is None:
            continue

        UR = create_uploader_resource(uploader)

        @ns.doc(
            description="Upload files received on `files[]` for processing",
            params=default_params,
            responses={
                400: "File list is empty or files are not on 'files[]' key",
                402: "Quota exceeded",
                403: "Missing `Tenant` or `Authorization` information",
                500: "Something went wrong while handling the request",
            },
        )
        @ns.expect(file_upload_parser)
        @ns.response(
            code=200, description="Upload response", model=file_validation_resp_model
        )
        class TempUploaderResource(UR):
            ...

        # Adding extra swagger query parameters if provided
        if uploader.query_params_on_upload:

            if isinstance(uploader.query_params_on_upload, list):
                params_dict = {}
                for param in uploader.query_params_on_upload:
                    params_dict[param["id"]] = {
                        "description": param["description"],
                        "allowed_values": param["allowed_values"],
                    }
            else:
                params_dict = {}
                for (
                    param_name,
                    param_description,
                ) in uploader.query_params_on_upload.items():
                    params_dict[param_name] = {"description": param_description}

            TempUploaderResource.__apidoc__.update({"params": {**default_params, **params_dict}})

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

        ns.add_resource(UploaderResource, uploader.upload_path)

    return ns

Classes

class FileUploadDetailedValidationSchema (*, only: Union[Sequence[str], Set[str], None] = None, exclude: Union[Sequence[str], Set[str]] = (), many: bool = False, context: Optional[Dict[~KT, ~VT]] = None, load_only: Union[Sequence[str], Set[str]] = (), dump_only: Union[Sequence[str], Set[str]] = (), partial: Union[bool, Sequence[str], Set[str]] = False, unknown: Optional[str] = None)

Base schema class with which to define custom schemas.

Example usage:

.. code-block:: python

import datetime as dt
from dataclasses import dataclass

from marshmallow import Schema, fields


@dataclass
class Album:
    title: str
    release_date: dt.date


class AlbumSchema(Schema):
    title = fields.Str()
    release_date = fields.Date()


album = Album("Beggars Banquet", dt.date(1968, 12, 6))
schema = AlbumSchema()
data = schema.dump(album)
data  # {'release_date': '1968-12-06', 'title': 'Beggars Banquet'}

:param only: Whitelist of the declared fields to select when instantiating the Schema. If None, all fields are used. Nested fields can be represented with dot delimiters. :param exclude: Blacklist of the declared fields to exclude when instantiating the Schema. If a field appears in both only and exclude, it is not used. Nested fields can be represented with dot delimiters. :param many: Should be set to True if obj is a collection so that the object will be serialized to a list. :param context: Optional context passed to :class:fields.Method and :class:fields.Function fields. :param load_only: Fields to skip during serialization (write-only fields) :param dump_only: Fields to skip during deserialization (read-only fields) :param partial: Whether to ignore missing fields and not require any fields declared. Propagates down to Nested fields as well. If its value is an iterable, only missing fields listed in that iterable will be ignored. Use dot delimiters to specify nested fields. :param unknown: Whether to exclude, include, or raise an error for unknown fields in the data. Use EXCLUDE, INCLUDE or RAISE.

Changed in version: 3.0.0

prefix parameter removed.

Changed in version: 2.0.0

__validators__, __preprocessors__, and __data_handlers__ are removed in favor of marshmallow.decorators.validates_schema, marshmallow.decorators.pre_load and marshmallow.decorators.post_dump. __accessor__ and __error_handler__ are deprecated. Implement the handle_error and get_attribute methods instead.

Expand source code

class FileUploadDetailedValidationSchema(Schema):
    status = fields.Str()
    message = fields.Str()
    filename = fields.Str()
    filepath = fields.Str()

Ancestors

marshmallow.schema.Schema
marshmallow.base.SchemaABC

Class variables

var filename
var filepath
var message
var opts
var status

class FileUploadEventDataSchema (*, only: Union[Sequence[str], Set[str], None] = None, exclude: Union[Sequence[str], Set[str]] = (), many: bool = False, context: Optional[Dict[~KT, ~VT]] = None, load_only: Union[Sequence[str], Set[str]] = (), dump_only: Union[Sequence[str], Set[str]] = (), partial: Union[bool, Sequence[str], Set[str]] = False, unknown: Optional[str] = None)

Base schema class with which to define custom schemas.

Example usage:

.. code-block:: python

import datetime as dt
from dataclasses import dataclass

from marshmallow import Schema, fields


@dataclass
class Album:
    title: str
    release_date: dt.date


class AlbumSchema(Schema):
    title = fields.Str()
    release_date = fields.Date()


album = Album("Beggars Banquet", dt.date(1968, 12, 6))
schema = AlbumSchema()
data = schema.dump(album)
data  # {'release_date': '1968-12-06', 'title': 'Beggars Banquet'}

Changed in version: 3.0.0

prefix parameter removed.

Changed in version: 2.0.0

Expand source code

class FileUploadEventDataSchema(Schema):
    tenant_id = fields.Str()
    uploader_id = fields.Str()
    event_id = fields.Str()
    filepaths = fields.List(fields.Str)
    flask_request = fields.List(fields.Raw)
    validation_response = fields.Nested(FileUploadValidationSchema)

Ancestors

marshmallow.schema.Schema
marshmallow.base.SchemaABC

Class variables

var event_id
var filepaths
var flask_request
var opts
var tenant_id
var uploader_id
var validation_response

class FileUploadRespSchema (*, only: Union[Sequence[str], Set[str], None] = None, exclude: Union[Sequence[str], Set[str]] = (), many: bool = False, context: Optional[Dict[~KT, ~VT]] = None, load_only: Union[Sequence[str], Set[str]] = (), dump_only: Union[Sequence[str], Set[str]] = (), partial: Union[bool, Sequence[str], Set[str]] = False, unknown: Optional[str] = None)

Base schema class with which to define custom schemas.

Example usage:

.. code-block:: python

import datetime as dt
from dataclasses import dataclass

from marshmallow import Schema, fields


@dataclass
class Album:
    title: str
    release_date: dt.date


class AlbumSchema(Schema):
    title = fields.Str()
    release_date = fields.Date()


album = Album("Beggars Banquet", dt.date(1968, 12, 6))
schema = AlbumSchema()
data = schema.dump(album)
data  # {'release_date': '1968-12-06', 'title': 'Beggars Banquet'}

Changed in version: 3.0.0

prefix parameter removed.

Changed in version: 2.0.0

Expand source code

class FileUploadRespSchema(Schema):
    status = fields.Str()
    message = fields.Str()
    #event_data = fields.List(fields.Nested(FileUploadEventDataSchema))
    event_id = fields.Str()

Ancestors

marshmallow.schema.Schema
marshmallow.base.SchemaABC

Class variables

var event_id
var message
var opts
var status

class FileUploadValidationSchema (*, only: Union[Sequence[str], Set[str], None] = None, exclude: Union[Sequence[str], Set[str]] = (), many: bool = False, context: Optional[Dict[~KT, ~VT]] = None, load_only: Union[Sequence[str], Set[str]] = (), dump_only: Union[Sequence[str], Set[str]] = (), partial: Union[bool, Sequence[str], Set[str]] = False, unknown: Optional[str] = None)

Base schema class with which to define custom schemas.

Example usage:

.. code-block:: python

import datetime as dt
from dataclasses import dataclass

from marshmallow import Schema, fields


@dataclass
class Album:
    title: str
    release_date: dt.date


class AlbumSchema(Schema):
    title = fields.Str()
    release_date = fields.Date()


album = Album("Beggars Banquet", dt.date(1968, 12, 6))
schema = AlbumSchema()
data = schema.dump(album)
data  # {'release_date': '1968-12-06', 'title': 'Beggars Banquet'}

Changed in version: 3.0.0

prefix parameter removed.

Changed in version: 2.0.0

Expand source code

class FileUploadValidationSchema(Schema):
    tenant_id = fields.Str()
    status = fields.Str()
    message = fields.Str()
    validation = fields.List(fields.Nested(FileUploadDetailedValidationSchema))

Ancestors

marshmallow.schema.Schema
marshmallow.base.SchemaABC

Class variables

var message
var opts
var status
var tenant_id
var validation