RESTful API - nemopss/mpt-kpi GitHub Wiki

from flask_restful import Resource, reqparse

docs = {}

class BasicResource(Resource):
    def get(self, source_id: int) -> dict:
        return {source_id: docs.get(source_id)}

    # FOR UPDATE
    def put(self, source_id: int):
        docs[source_id] = source_id
        return {source_id: source_id}

    # FOR CREATE
    def post(self, source_id: int):
        docs[source_id] = source_id
        return {source_id: source_id}, 201

from flask_restful import Resource, reqparse
from services.certificates_generator import Certificate, certificates_list

class CertificatesResource(Resource):
    def get(self) -> dict:
        return [certificate._asdict() for certificate in certificates_list]

from flask_restful import Resource, reqparse
from services.criteries_generator import Criterion, criteries_list

class CriteriesResource(Resource):
    def get(self) -> dict:
        return [criterion._asdict() for criterion in criteries_list]

from flask_restful import Resource, reqparse
from flask import request
from services.employees_generator import Employee, employees_list

# parser = reqparse.RequestParser()
# (
#     parser.add_argument(argument) for argument in
#     'id first_name last_name surname job_id phone email'.split()
# )

class EmployeesResource(Resource):
    def get(self) -> dict:
        return [employee._asdict() for employee in employees_list]

    def post(self) -> dict:
        args = request.json
        #args = parser.parse_args()
        print(args)
        employee = Employee(
            id=args['id'],
            first_name=args['first_name'],
            last_name=args['last_name'],
            surname=args['surname'],
            job_id=args['job_id'],
            phone=args['phone'],
            email=args['email'],
        )
        employees_list.append(employee)
        return employees_list[-1]._asdict()

import os

from werkzeug.utils import secure_filename
from flask_restful import Resource, abort
from flask import request, send_from_directory

ALLOWED_EXTENSIONS = ['pdf', 'doc', 'docx', 'png']

def find_local_file(target_name) -> bool:
    print(target_name)
    for (root, folder, file_name) in os.walk('docs'):
        if len(file_name) == 0: return False
        if target_name in file_name: return True
    return False

def validate_extension(file_ext) -> bool:
    for ext in ALLOWED_EXTENSIONS:
        if file_ext == ext: return True
    return False

def build_name(file_name, file_ext) -> str: return file_name + '.' + file_ext

class PDFResource(Resource):
    def post(self):
        file = request.files.get('image')
        print(file.filename)
        save_filename, save_ext = secure_filename(file.filename).split('.')
        if not validate_extension(save_ext): abort(415)
        if find_local_file(file.filename):
            save_filename += '_copy'
        save_filename = build_name(save_filename, save_ext)
        file.save(dst="docs/"+secure_filename(save_filename))
        return {'file': 'saved:OK'}, 201

    def get(self, filename):
        if find_local_file(filename):
            print(f'FOUND {filename}')
            # Возвращает файл, который открывается в браузере
            # На фронте можно просто дать ссылку на docs/<имя файла>
            # <a href="docs/filo.pdf">File link</a>, должно работать
            return send_from_directory('docs', secure_filename(filename))
        else:
            abort(404)

    def delete(self, filename):
        try:
            os.remove(os.path.join('docs', secure_filename(filename)))
            return {'file': 'deleted:OK'}, 200
        except WindowsError as e:
            print(e)
            abort(404)