jedoist-client/src/main.py

import falcon
import json
import os
from typing import Optional

# https://falcon.readthedocs.io/en/stable/user/quickstart.html#learning-by-example

class AuthMiddleware:
    def process_request(self, req, resp):
        token = req.get_header('Authorization')
        account_id = req.get_header('Account-ID')

        challenges = ['Token type="Fernet"']

        if token is None:
            description = 'Please provide an auth token as part of the request.'

            raise falcon.HTTPUnauthorized(
                title='Auth token required',
                description=description,
                challenges=challenges,
                href='http://docs.example.com/auth',
            )

        if not self._token_is_valid(token, account_id):
            description = (
                'The provided auth token is not valid. '
                'Please request a new token and try again.'
            )

            raise falcon.HTTPUnauthorized(
                title='Authentication required',
                description=description,
                challenges=challenges,
                href='http://docs.example.com/auth',
            )

    def _token_is_valid(self, token, account_id):
        return os.environ["TOKEN"]  # Suuuuuure it's valid...


class RequireJSON:
    def process_request(self, req, resp):
        if not req.client_accepts_json:
            raise falcon.HTTPNotAcceptable(
                description='This API only supports responses encoded as JSON.',
                href='http://docs.examples.com/api/json',
            )

        if req.method in ('POST', 'PUT'):
            if req.content_type is None or 'application/json' not in req.content_type:
                raise falcon.HTTPUnsupportedMediaType(
                    title='This API only supports requests encoded as JSON.',
                    href='http://docs.examples.com/api/json',
                )


class JSONTranslator:
    # NOTE: Normally you would simply use req.media and resp.media for
    # this particular use case; this example serves only to illustrate
    # what is possible.

    def process_request(self, req, resp):
        # req.stream corresponds to the WSGI wsgi.input environ variable,
        # and allows you to read bytes from the request body.
        #
        # See also: PEP 3333
        if req.content_length in (None, 0):
            # Nothing to do
            return

        body = req.stream.read()
        if not body:
            raise falcon.HTTPBadRequest(
                title='Empty request body',
                description='A valid JSON document is required.',
            )

        try:
            req.context.doc = json.loads(body.decode('utf-8'))

        except (ValueError, UnicodeDecodeError):
            description = (
                'Could not decode the request body. The '
                'JSON was incorrect or not encoded as '
                'UTF-8.'
            )

            raise falcon.HTTPBadRequest(title='Malformed JSON', description=description)

    def process_response(self, req, resp, resource, req_succeeded):
        if not hasattr(resp.context, 'result'):
            return

        resp.text = json.dumps(resp.context.result)


def max_body(limit):
    def hook(req, resp, resource, params):
        length = req.content_length
        if length is not None and length > limit:
            msg = (
                'The size of the request is too large. The body must not '
                'exceed ' + str(limit) + ' bytes in length.'
            )

            raise falcon.HTTPPayloadTooLarge(
                title='Request body is too large', description=msg
            )

    return hook

########################################################################################################

class CreateTaskResource:

    TODO_HEADER = '# ToDo'

    def __init__(self, config_file: str, todos_path: str) -> None:
        self.__todos_path = todos_path
        self.__config = self.__read_config_file(config_file)

    def on_post(self, req, resp):
        params = req.params
        list_id = 'default'
        if 'list' in params:
            list_id = params['list']
        file_path = self.__detect_todo_file_path(list_id)
        if file_path == None:
            quote = {
                'title': 'Ошибка настройки',
                'description': 'В URL не указан параметр "list"'
            }
            resp.media = quote
            resp.status = falcon.HTTP_500
        else:
            task = req.context.doc['task']
            self.__create_task(task, file_path)
            quote = {
                'title': 'Получена задача',
                'description': task
            }
            resp.media = quote
            resp.status = falcon.HTTP_201

    def __read_config_file(self, config_file: str) -> None:
        with open(config_file, 'r') as f:
            return json.load(f)

    def __detect_todo_file_path(self, todo_list_id) -> Optional[str]:
        if not 'lists' in self.__config or not todo_list_id in self.__config['lists']:
            return None
        return "{0}/{1}".format(self.__todos_path, self.__config['lists'][todo_list_id])

    def __create_task(self, text: str, file_path: str) -> None:
        content = []
        section_found = False
        is_waiting_empty_line = False
        with open(file_path, 'r', encoding='UTF-8') as file:
            while line := file.readline():
                if not section_found:
                    if line.strip() == CreateTaskResource.TODO_HEADER:
                        section_found = True
                        is_waiting_empty_line = True
                content.append(line)
                if section_found and is_waiting_empty_line == True and line.strip() == '':
                    content.append("- [ ] {0}\n".format(text))
                    is_waiting_empty_line = False
        with open(file_path, 'w', encoding='utf-8') as file:
            file.writelines(content)

########################################################################################################

app = falcon.App(
    middleware=[
        AuthMiddleware(),
        RequireJSON(),
        JSONTranslator()
    ]
)
app.add_route('/create-task', CreateTaskResource(config_file=os.environ['CONFIG_FILE'],
                                                 todos_path=os.environ['TODOS_PATH']))