validata_util.py 1.39 KB
Newer Older
1 2 3
""" Call validation code """

import logging
4
from abc import abstractmethod, ABC
5
from io import BytesIO
6
from pathlib import Path
7

8 9 10
log = logging.getLogger(__name__)


11 12
class ValidataResource(ABC):
    """A resource to validate: url or uploaded file"""
13

14
    def __init__(self, type_):
Pierre Dittgen's avatar
wip  
Pierre Dittgen committed
15
        self.type = type_
16

17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33
    @abstractmethod
    def build_tabulator_stream_args(self):
        """return (source, option_dict)"""
        pass


class URLValidataResource(ValidataResource):
    """URL resource"""

    def __init__(self, url):
        """Built from URL"""
        super().__init__('url')
        self.url = url

    def build_tabulator_stream_args(self):
        """URL implementation"""
        return (self.url, {})
34 35


36 37
class UploadedFileValidataResource(ValidataResource):
    """Uploaded file resource"""
Pierre Dittgen's avatar
wip  
Pierre Dittgen committed
38

39 40 41 42 43
    def __init__(self, filename, bytes_content):
        """Built from file name and content"""
        super().__init__('file')
        self.filename = filename
        self.content = bytes_content
Pierre Dittgen's avatar
wip  
Pierre Dittgen committed
44

45 46
    def build_reader(self):
        return BytesIO(self.content)
47

48 49 50 51 52
    def __detect_format(self):
        ext = Path(self.filename).suffix
        if ext in ('.csv', '.tsv', '.ods', '.xls', '.xlsx'):
            return ext[1:]
        return None
53

54 55 56 57
    def build_tabulator_stream_args(self):
        """Uploaded file implementation"""
        return (self.build_reader(), {
            'format': self.__detect_format()})