lab_1 #3
@ -1,52 +0,0 @@
|
||||
import importlib
|
||||
import os
|
||||
import traceback
|
||||
|
||||
import matplotlib
|
||||
from apiflask import APIBlueprint, APIFlask
|
||||
from flask_cors import CORS
|
||||
|
||||
matplotlib.use("agg")
|
||||
|
||||
cors = CORS()
|
||||
api_bp = APIBlueprint("api", __name__, url_prefix="/api/v1")
|
||||
dataset_path: str | None = None
|
||||
|
||||
|
||||
class Config:
|
||||
SECRET_KEY = "secret!"
|
||||
SEND_FILE_MAX_AGE_DEFAULT = -1
|
||||
|
||||
|
||||
def create_app():
|
||||
global dataset_path
|
||||
|
||||
# Create and configure app
|
||||
app = APIFlask(
|
||||
"MAI Service",
|
||||
title="MAI Service API",
|
||||
docs_path="/",
|
||||
version="1.0",
|
||||
static_folder="",
|
||||
template_folder="",
|
||||
)
|
||||
app.config.from_object(Config)
|
||||
|
||||
dataset_path = os.path.join(app.instance_path, "dataset")
|
||||
os.makedirs(dataset_path, exist_ok=True)
|
||||
|
||||
@app.errorhandler(Exception)
|
||||
def my_error_processor(error):
|
||||
traceback.print_exception(error)
|
||||
return {"message": str(error), "detail": "No details"}, 500
|
||||
|
||||
# Import custom REST methods
|
||||
importlib.import_module("backend.api")
|
||||
|
||||
# Enable REST API
|
||||
app.register_blueprint(api_bp)
|
||||
|
||||
# Enable app extensions
|
||||
cors.init_app(app)
|
||||
|
||||
return app
|
Binary file not shown.
Binary file not shown.
Binary file not shown.
@ -1,57 +0,0 @@
|
||||
from apiflask import FileSchema, Schema, fields
|
||||
from flask import send_file
|
||||
|
||||
from backend import api_bp, dataset_path
|
||||
from backend.service import Service
|
||||
|
||||
|
||||
class FileUpload(Schema):
|
||||
file = fields.File(required=True)
|
||||
|
||||
|
||||
class ColumnInfoDto(Schema):
|
||||
datatype = fields.String()
|
||||
items = fields.List(fields.String())
|
||||
|
||||
|
||||
class TableColumnDto(Schema):
|
||||
name = fields.String()
|
||||
datatype = fields.String()
|
||||
items = fields.List(fields.String())
|
||||
|
||||
|
||||
service = Service(dataset_path)
|
||||
|
||||
|
||||
@api_bp.post("/dataset")
|
||||
@api_bp.input(FileUpload, location="files")
|
||||
def upload_dataset(files_data):
|
||||
uploaded_file = files_data["file"]
|
||||
return service.upload_dataset(uploaded_file)
|
||||
|
||||
|
||||
@api_bp.get("/dataset")
|
||||
def get_all_datasets():
|
||||
return service.get_all_datasets()
|
||||
|
||||
|
||||
@api_bp.get("/dataset/<string:name>")
|
||||
@api_bp.output(TableColumnDto(many=True))
|
||||
def get_dataset_info(name: str):
|
||||
return service.get_dataset_info(name)
|
||||
|
||||
|
||||
@api_bp.get("/dataset/<string:name>/<string:column>")
|
||||
@api_bp.output(ColumnInfoDto)
|
||||
def get_column_info(name: str, column: str):
|
||||
return service.get_column_info(name, column)
|
||||
|
||||
|
||||
@api_bp.get("/dataset/draw/hist/<string:name>/<string:column>")
|
||||
@api_bp.output(
|
||||
FileSchema(type="string", format="binary"), content_type="image/png", example=""
|
||||
)
|
||||
def get_dataset_hist(name: str, column: str):
|
||||
data = service.get_hist(name, column)
|
||||
data.seek(0)
|
||||
return send_file(data, download_name=f"{name}.hist.png", mimetype="image/png")
|
@ -1,59 +0,0 @@
|
||||
import io
|
||||
import os
|
||||
import pathlib
|
||||
from typing import BinaryIO, Dict, List
|
||||
|
||||
import pandas as pd
|
||||
from matplotlib.figure import Figure
|
||||
from werkzeug.datastructures import FileStorage
|
||||
from werkzeug.utils import secure_filename
|
||||
|
||||
|
||||
class Service:
|
||||
def __init__(self, dataset_path: str | None) -> None:
|
||||
if dataset_path is None:
|
||||
raise Exception("Dataset path is not defined")
|
||||
self.__path: str = dataset_path
|
||||
|
||||
def __get_dataset(self, filename: str) -> pd.DataFrame:
|
||||
full_file_name = os.path.join(self.__path, secure_filename(filename))
|
||||
return pd.read_csv(full_file_name)
|
||||
|
||||
def upload_dataset(self, file: FileStorage) -> str:
|
||||
if file.filename is None:
|
||||
raise Exception("Dataset upload error")
|
||||
file_name: str = file.filename
|
||||
full_file_name = os.path.join(self.__path, secure_filename(file_name))
|
||||
file.save(full_file_name)
|
||||
return file_name
|
||||
|
||||
def get_all_datasets(self) -> List[str]:
|
||||
return [file.name for file in pathlib.Path(self.__path).glob("*.csv")]
|
||||
|
||||
def get_dataset_info(self, filename) -> List[Dict]:
|
||||
dataset = self.__get_dataset(filename)
|
||||
dataset_info = []
|
||||
for column in dataset.columns:
|
||||
items = dataset[column].astype(str)
|
||||
column_info = {
|
||||
"name": column,
|
||||
"datatype": dataset.dtypes[column],
|
||||
"items": items,
|
||||
}
|
||||
dataset_info.append(column_info)
|
||||
return dataset_info
|
||||
|
||||
def get_column_info(self, filename, column) -> Dict:
|
||||
dataset = self.__get_dataset(filename)
|
||||
datatype = dataset.dtypes[column]
|
||||
items = sorted(dataset[column].astype(str).unique())
|
||||
return {"datatype": datatype, "items": items}
|
||||
|
||||
def get_hist(self, filename, column) -> BinaryIO:
|
||||
dataset = self.__get_dataset(filename)
|
||||
bytes = io.BytesIO()
|
||||
plot: Figure | None = dataset.plot.hist(column=[column], bins=80).get_figure()
|
||||
if plot is None:
|
||||
raise Exception("Can't create hist plot")
|
||||
plot.savefig(bytes, dpi=300, format="png")
|
||||
return bytes
|
File diff suppressed because it is too large
Load Diff
Loading…
Reference in New Issue
Block a user