mirror of
https://github.com/shadoll/scatcher.git
synced 2025-12-20 06:29:59 +00:00
Add schemas
This commit is contained in:
58
manager/history.py
Normal file
58
manager/history.py
Normal file
@@ -0,0 +1,58 @@
|
|||||||
|
import json
|
||||||
|
import os
|
||||||
|
from schema.methods import Methods
|
||||||
|
|
||||||
|
class History:
|
||||||
|
HISTORY_LIMIT = 10
|
||||||
|
HISTORY_STORAGE = "storage"
|
||||||
|
|
||||||
|
def __init__(self, namespace: str = "requests") -> None:
|
||||||
|
self.namespace = namespace
|
||||||
|
self.filename = f"{self.HISTORY_STORAGE}/{namespace}.json"
|
||||||
|
self.data: list = []
|
||||||
|
self.check_directory()
|
||||||
|
|
||||||
|
def check_directory(self) -> None:
|
||||||
|
if not os.path.exists(self.HISTORY_STORAGE):
|
||||||
|
os.makedirs(self.HISTORY_STORAGE)
|
||||||
|
|
||||||
|
def load(self) -> None:
|
||||||
|
try:
|
||||||
|
with open(self.filename, "r") as f:
|
||||||
|
self.data = json.load(f)
|
||||||
|
except (FileNotFoundError, json.JSONDecodeError):
|
||||||
|
self.data = []
|
||||||
|
|
||||||
|
def save(self) -> None:
|
||||||
|
with open(self.filename, "w") as f:
|
||||||
|
json.dump(self.data, f, indent=4)
|
||||||
|
|
||||||
|
def clear(self) -> None:
|
||||||
|
self.data = []
|
||||||
|
self.save()
|
||||||
|
|
||||||
|
def all(self) -> list | None:
|
||||||
|
self.load()
|
||||||
|
if len(self.data) == 0:
|
||||||
|
return None
|
||||||
|
for item in self.data:
|
||||||
|
if "method" in item:
|
||||||
|
item["method"] = Methods(item["method"])
|
||||||
|
return self.data
|
||||||
|
|
||||||
|
def get(self, index: int) -> dict | None:
|
||||||
|
self.all()
|
||||||
|
|
||||||
|
return self.data[index] if index < len(self.data) else None
|
||||||
|
|
||||||
|
def last(self) -> dict | None:
|
||||||
|
self.all()
|
||||||
|
|
||||||
|
return self.data[-1] if len(self.data) > 0 else None
|
||||||
|
|
||||||
|
def add(self, request_data: dict) -> None:
|
||||||
|
self.load()
|
||||||
|
self.data.append(request_data)
|
||||||
|
if len(self.data) > self.HISTORY_LIMIT:
|
||||||
|
self.data.pop(0)
|
||||||
|
self.save()
|
||||||
7
schema/answer.py
Normal file
7
schema/answer.py
Normal file
@@ -0,0 +1,7 @@
|
|||||||
|
from pydantic import BaseModel
|
||||||
|
from schema.status import Status
|
||||||
|
|
||||||
|
|
||||||
|
class Answer(BaseModel):
|
||||||
|
status: Status
|
||||||
|
message: str
|
||||||
11
schema/methods.py
Normal file
11
schema/methods.py
Normal file
@@ -0,0 +1,11 @@
|
|||||||
|
from enum import Enum
|
||||||
|
|
||||||
|
|
||||||
|
class Methods(str, Enum):
|
||||||
|
GET = "GET"
|
||||||
|
POST = "POST"
|
||||||
|
PUT = "PUT"
|
||||||
|
DELETE = "DELETE"
|
||||||
|
PATCH = "PATCH"
|
||||||
|
OPTIONS = "OPTIONS"
|
||||||
|
HEAD = "HEAD"
|
||||||
10
schema/request_data.py
Normal file
10
schema/request_data.py
Normal file
@@ -0,0 +1,10 @@
|
|||||||
|
from pydantic import BaseModel
|
||||||
|
from schema.methods import Methods
|
||||||
|
|
||||||
|
|
||||||
|
class RequestData(BaseModel):
|
||||||
|
data: dict
|
||||||
|
method: Methods = Methods.GET
|
||||||
|
url: str
|
||||||
|
headers: dict
|
||||||
|
time: str
|
||||||
7
schema/status.py
Normal file
7
schema/status.py
Normal file
@@ -0,0 +1,7 @@
|
|||||||
|
from enum import Enum
|
||||||
|
|
||||||
|
|
||||||
|
class Status(str, Enum):
|
||||||
|
ok = "ok"
|
||||||
|
success = "success"
|
||||||
|
error = "error"
|
||||||
Reference in New Issue
Block a user