mirror of
https://github.com/lrnselfreliance/wrolpi.git
synced 2026-10-26 10:51:36 -04:00
Upgrading Sanic. Updating tests and schemas.
This commit is contained in:
Roland
parent
0bd5d17b46
commit
b1ed453466
18 changed files with 353 additions and 592 deletions
|
|
@ -3,11 +3,12 @@ from http import HTTPStatus
|
|||
|
||||
from requests import Request
|
||||
from sanic import response
|
||||
from sanic_ext import validate
|
||||
from sanic_ext.extensions.openapi import openapi
|
||||
|
||||
from wrolpi.common import logger, wrol_mode_check, api_param_limiter
|
||||
from wrolpi.errors import ValidationError
|
||||
from wrolpi.root_api import get_blueprint, json_response
|
||||
from wrolpi.schema import validate_doc, JSONErrorResponse
|
||||
from wrolpi.schema import JSONErrorResponse
|
||||
from . import lib
|
||||
from .schema import ArchiveSearchRequest, ArchiveSearchResponse
|
||||
|
||||
|
|
@ -19,12 +20,8 @@ logger = logger.getChild(__name__)
|
|||
|
||||
|
||||
@bp.delete('/<archive_id:int>')
|
||||
@validate_doc(
|
||||
'Delete an individual archive',
|
||||
responses=[
|
||||
(HTTPStatus.NOT_FOUND, JSONErrorResponse),
|
||||
]
|
||||
)
|
||||
@openapi.description('Delete an individual archive')
|
||||
@openapi.response(HTTPStatus.NOT_FOUND, JSONErrorResponse)
|
||||
@wrol_mode_check
|
||||
async def delete_archive(_: Request, archive_id: int):
|
||||
lib.delete_archive(archive_id)
|
||||
|
|
@ -49,20 +46,15 @@ archive_offset_limiter = api_param_limiter(100, 0)
|
|||
|
||||
|
||||
@bp.post('/search')
|
||||
@validate_doc(
|
||||
'Search archive contents and titles',
|
||||
consumes=ArchiveSearchRequest,
|
||||
produces=ArchiveSearchResponse,
|
||||
optional_body=True,
|
||||
)
|
||||
async def search_archives(_: Request, data: dict):
|
||||
try:
|
||||
search_str = data.get('search_str')
|
||||
domain = data.get('domain')
|
||||
limit = archive_limit_limiter(data.get('limit'))
|
||||
offset = archive_offset_limiter(data.get('offset'))
|
||||
except Exception as e:
|
||||
raise ValidationError('Unable to validate search queries') from e
|
||||
@openapi.description('Search archive contents and titles')
|
||||
@openapi.response(HTTPStatus.OK, ArchiveSearchResponse)
|
||||
@openapi.response(HTTPStatus.NOT_FOUND, JSONErrorResponse)
|
||||
@validate(ArchiveSearchRequest)
|
||||
async def search_archives(_: Request, body: ArchiveSearchRequest):
|
||||
search_str = body.search_str
|
||||
domain = body.domain
|
||||
limit = archive_limit_limiter(body.limit)
|
||||
offset = archive_offset_limiter(body.offset)
|
||||
|
||||
archives, total = lib.search(search_str, domain, limit, offset)
|
||||
ret = dict(archives=archives, totals=dict(archives=total))
|
||||
|
|
|
|||
|
|
@ -1,31 +1,37 @@
|
|||
from sanic_openapi import doc
|
||||
from dataclasses import dataclass
|
||||
from datetime import datetime
|
||||
from typing import List, Optional
|
||||
|
||||
|
||||
@dataclass
|
||||
class ArchiveDict:
|
||||
id = doc.Integer()
|
||||
url_id = doc.Integer()
|
||||
domain_id = doc.Integer()
|
||||
singlefile_path = doc.String()
|
||||
readability_path = doc.String()
|
||||
readability_json_path = doc.String()
|
||||
readability_txt_path = doc.String()
|
||||
screenshot_path = doc.String()
|
||||
title = doc.String()
|
||||
archive_datetime = doc.DateTime()
|
||||
id: int
|
||||
url_id: int
|
||||
domain_id: int
|
||||
singlefile_path: str
|
||||
readability_path: str
|
||||
readability_json_path: str
|
||||
readability_txt_path: str
|
||||
screenshot_path: str
|
||||
title: str
|
||||
archive_datetime: datetime
|
||||
|
||||
|
||||
@dataclass
|
||||
class DomainDict:
|
||||
id = doc.Integer()
|
||||
domain = doc.String()
|
||||
id: int
|
||||
domain: str
|
||||
|
||||
|
||||
@dataclass
|
||||
class ArchiveSearchRequest:
|
||||
search_str = doc.String()
|
||||
domain = doc.String()
|
||||
offset = doc.Integer()
|
||||
limit = doc.Integer()
|
||||
search_str: Optional[str] = None
|
||||
domain: Optional[str] = None
|
||||
offset: Optional[int] = None
|
||||
limit: Optional[int] = None
|
||||
|
||||
|
||||
@dataclass
|
||||
class ArchiveSearchResponse:
|
||||
videos = doc.List(ArchiveDict)
|
||||
totals = doc.Dictionary()
|
||||
videos: List[ArchiveDict]
|
||||
totals: dict
|
||||
|
|
|
|||
|
|
@ -58,7 +58,7 @@ def test_archives_search_no_query(archive_factory, test_client):
|
|||
archive_factory('example.com')
|
||||
|
||||
# All archives are returned when no `search_str` is passed.
|
||||
request, response = test_client.post('/api/archive/search')
|
||||
request, response = test_client.post('/api/archive/search', content='{}')
|
||||
assert response.status_code == HTTPStatus.OK, response.json
|
||||
assert [i['id'] for i in response.json['archives']] == list(range(100, 80, -1))
|
||||
assert response.json['totals']['archives'] == 100
|
||||
|
|
|
|||
|
|
@ -2,11 +2,12 @@ from http import HTTPStatus
|
|||
|
||||
from sanic import response
|
||||
from sanic.request import Request
|
||||
from sanic_ext import validate
|
||||
from sanic_ext.extensions.openapi import openapi
|
||||
|
||||
from wrolpi.common import run_after, remove_dict_value_whitespace
|
||||
from wrolpi.errors import ValidationError
|
||||
from wrolpi.root_api import get_blueprint, json_response
|
||||
from wrolpi.schema import validate_doc
|
||||
from .common import get_inventory_by_category, get_inventory_by_subcategory, get_inventory_by_name, \
|
||||
save_inventories_file
|
||||
from .inventory import get_items, save_item, delete_items, \
|
||||
|
|
@ -45,10 +46,8 @@ def inventory_get(_: Request, inventory_id: int):
|
|||
|
||||
|
||||
@bp.post('/')
|
||||
@validate_doc(
|
||||
'Save a new inventory',
|
||||
consumes=InventoryPostRequest,
|
||||
)
|
||||
@openapi.description('Save a new inventory')
|
||||
@validate(InventoryPostRequest)
|
||||
@run_after(save_inventories_file)
|
||||
def post_inventory(_: Request, data: dict):
|
||||
# Cleanup the whitespace.
|
||||
|
|
@ -59,10 +58,8 @@ def post_inventory(_: Request, data: dict):
|
|||
|
||||
|
||||
@bp.put('/<inventory_id:int>')
|
||||
@validate_doc(
|
||||
'Update an inventory',
|
||||
consumes=InventoryPutRequest,
|
||||
)
|
||||
@openapi.description('Update an inventory')
|
||||
@validate(InventoryPutRequest)
|
||||
@run_after(save_inventories_file)
|
||||
def put_inventory(_: Request, inventory_id: int, data: dict):
|
||||
# Cleanup the whitespace.
|
||||
|
|
@ -73,6 +70,7 @@ def put_inventory(_: Request, inventory_id: int, data: dict):
|
|||
|
||||
|
||||
@bp.delete('/<inventory_id:int>')
|
||||
@openapi.description('Delete an inventory.')
|
||||
@run_after(save_inventories_file)
|
||||
def inventory_delete(_: Request, inventory_id: int):
|
||||
delete_inventory(inventory_id)
|
||||
|
|
@ -80,19 +78,15 @@ def inventory_delete(_: Request, inventory_id: int):
|
|||
|
||||
|
||||
@bp.get('/<inventory_id:int>/item')
|
||||
@validate_doc(
|
||||
"Get all items from an inventory.",
|
||||
)
|
||||
@openapi.description('Get all items from an inventory.')
|
||||
def items_get(_: Request, inventory_id: int):
|
||||
items = get_items(inventory_id)
|
||||
return json_response({'items': items})
|
||||
|
||||
|
||||
@bp.post('/<inventory_id:int>/item')
|
||||
@validate_doc(
|
||||
"Save an item into it's inventory.",
|
||||
consumes=ItemPostRequest,
|
||||
)
|
||||
@openapi.description("Save an item into it's inventory.")
|
||||
@validate(ItemPostRequest)
|
||||
@run_after(save_inventories_file)
|
||||
def post_item(_: Request, inventory_id: int, data: dict):
|
||||
# Cleanup the whitespace.
|
||||
|
|
@ -103,10 +97,8 @@ def post_item(_: Request, inventory_id: int, data: dict):
|
|||
|
||||
|
||||
@bp.put('/item/<item_id:int>')
|
||||
@validate_doc(
|
||||
"Update an item",
|
||||
consumes=ItemPutRequest,
|
||||
)
|
||||
@openapi.description("Update an item.")
|
||||
@validate(ItemPutRequest)
|
||||
@run_after(save_inventories_file)
|
||||
def put_item(_: Request, item_id: int, data: dict):
|
||||
# Cleanup the whitespace.
|
||||
|
|
@ -118,9 +110,7 @@ def put_item(_: Request, item_id: int, data: dict):
|
|||
|
||||
@bp.delete('/item/<item_ids:[0-9,]+>')
|
||||
@bp.delete('/item/<item_ids:int>')
|
||||
@validate_doc(
|
||||
'Delete items from an inventory.',
|
||||
)
|
||||
@openapi.description('Delete items from an inventory.')
|
||||
@run_after(save_inventories_file)
|
||||
def item_delete(_: Request, item_ids: str):
|
||||
try:
|
||||
|
|
|
|||
|
|
@ -1,37 +1,43 @@
|
|||
from sanic_openapi import doc
|
||||
from dataclasses import dataclass
|
||||
from datetime import date, datetime
|
||||
from decimal import Decimal
|
||||
|
||||
|
||||
@dataclass
|
||||
class ItemPostRequest:
|
||||
brand = doc.String()
|
||||
name = doc.String(required=True)
|
||||
item_size = doc.Float(required=True)
|
||||
unit = doc.String(required=True)
|
||||
count = doc.Float(required=True)
|
||||
category = doc.String()
|
||||
subcategory = doc.String()
|
||||
expiration_date = doc.Date()
|
||||
brand: str
|
||||
name: str
|
||||
item_size: Decimal
|
||||
unit: str
|
||||
count: Decimal
|
||||
category: str
|
||||
subcategory: str
|
||||
expiration_date: date
|
||||
|
||||
|
||||
@dataclass
|
||||
class ItemPutRequest:
|
||||
id = doc.Integer() # Ignored in favor of URL
|
||||
brand = doc.String()
|
||||
name = doc.String(required=True)
|
||||
item_size = doc.Float(required=True)
|
||||
unit = doc.String(required=True)
|
||||
count = doc.Float(required=True)
|
||||
serving = doc.Float()
|
||||
category = doc.String()
|
||||
subcategory = doc.String()
|
||||
expiration_date = doc.Date()
|
||||
purchase_date = doc.Date()
|
||||
created_at = doc.DateTime()
|
||||
deleted_at = doc.DateTime()
|
||||
inventory_id = doc.Integer(required=True)
|
||||
id: int # Ignored in favor of URL
|
||||
brand: str
|
||||
name: str
|
||||
item_size: Decimal
|
||||
unit: str
|
||||
count: Decimal
|
||||
serving: Decimal
|
||||
category: str
|
||||
subcategory: str
|
||||
expiration_date: date
|
||||
purchase_date: date
|
||||
created_at: datetime
|
||||
deleted_at: datetime
|
||||
inventory_id: int
|
||||
|
||||
|
||||
@dataclass
|
||||
class InventoryPostRequest:
|
||||
name = doc.String(required=True)
|
||||
name: str
|
||||
|
||||
|
||||
@dataclass
|
||||
class InventoryPutRequest:
|
||||
name = doc.String(required=True)
|
||||
name: str
|
||||
|
|
|
|||
|
|
@ -4,10 +4,11 @@ from pathlib import Path
|
|||
from urllib.parse import urlparse
|
||||
|
||||
from sanic import Blueprint, response
|
||||
from sanic_ext import validate
|
||||
from sanic_ext.extensions.openapi import openapi
|
||||
|
||||
from modules.map.schema import PBFPostRequest, PBFPostResponse
|
||||
from wrolpi.common import create_websocket_feed
|
||||
from wrolpi.schema import validate_doc
|
||||
|
||||
NAME = 'map'
|
||||
|
||||
|
|
@ -17,11 +18,9 @@ download_queue, download_event = create_websocket_feed('pbf_progress', '/feeds/p
|
|||
|
||||
|
||||
@api_bp.route('/pbf', methods=['POST'])
|
||||
@validate_doc(
|
||||
summary='Queue a PBF file for download and processing',
|
||||
consumes=PBFPostRequest,
|
||||
produces=PBFPostResponse,
|
||||
)
|
||||
@openapi.description('Queue a PBF file for download and processing')
|
||||
@openapi.response(HTTPStatus.OK, PBFPostResponse)
|
||||
@validate(PBFPostRequest)
|
||||
def pbf_post(request, data: dict):
|
||||
pbf_url = data.get('pbf_url')
|
||||
parsed = urlparse(pbf_url)
|
||||
|
|
|
|||
|
|
@ -1,9 +1,11 @@
|
|||
from sanic_openapi import doc
|
||||
from dataclasses import dataclass
|
||||
|
||||
|
||||
@dataclass
|
||||
class PBFPostRequest:
|
||||
pbf_url = doc.String()
|
||||
pbf_url: str
|
||||
|
||||
|
||||
@dataclass
|
||||
class PBFPostResponse:
|
||||
success = doc.String()
|
||||
success: str
|
||||
|
|
|
|||
|
|
@ -4,9 +4,10 @@ from string import ascii_letters
|
|||
|
||||
from sanic import response
|
||||
from sanic.request import Request
|
||||
from sanic_ext import validate
|
||||
from sanic_ext.extensions.openapi import openapi
|
||||
|
||||
from wrolpi.root_api import get_blueprint
|
||||
from wrolpi.schema import validate_doc
|
||||
from .common import generate_pdf, encrypt_otp, decrypt_otp
|
||||
from .schema import EncryptOTPRequest, DecryptOTPRequest
|
||||
|
||||
|
|
@ -14,20 +15,16 @@ bp = get_blueprint('OTP', '/api/otp')
|
|||
|
||||
|
||||
@bp.post('/encrypt_otp')
|
||||
@validate_doc(
|
||||
summary='Encrypt a message with OTP',
|
||||
consumes=EncryptOTPRequest,
|
||||
)
|
||||
@openapi.description('Encrypt a message with OTP.')
|
||||
@validate(EncryptOTPRequest)
|
||||
async def post_encrypt_otp(_: Request, data: dict):
|
||||
data = encrypt_otp(data['otp'], data['plaintext'])
|
||||
return response.json(data)
|
||||
|
||||
|
||||
@bp.post('/decrypt_otp')
|
||||
@validate_doc(
|
||||
summary='Decrypt a message with OTP',
|
||||
consumes=DecryptOTPRequest,
|
||||
)
|
||||
@openapi.description('Decrypt a message with OTP.')
|
||||
@validate(DecryptOTPRequest)
|
||||
async def post_decrypt_otp(_: Request, data: dict):
|
||||
data = decrypt_otp(data['otp'], data['ciphertext'])
|
||||
return response.json(data)
|
||||
|
|
|
|||
|
|
@ -1,11 +1,13 @@
|
|||
from sanic_openapi import doc
|
||||
from dataclasses import dataclass
|
||||
|
||||
|
||||
@dataclass
|
||||
class EncryptOTPRequest:
|
||||
otp = doc.String(required=True)
|
||||
plaintext = doc.String(required=True)
|
||||
otp: str
|
||||
plaintext: str
|
||||
|
||||
|
||||
@dataclass
|
||||
class DecryptOTPRequest:
|
||||
otp = doc.String(required=True)
|
||||
ciphertext = doc.String(required=True)
|
||||
otp: str
|
||||
ciphertext: str
|
||||
|
|
|
|||
|
|
@ -4,12 +4,14 @@ from http import HTTPStatus
|
|||
|
||||
from sanic import Blueprint, response
|
||||
from sanic.request import Request
|
||||
from sanic_ext import validate
|
||||
from sanic_ext.extensions.openapi import openapi
|
||||
|
||||
from wrolpi.common import create_websocket_feed, get_sanic_url, \
|
||||
wrol_mode_check
|
||||
from wrolpi.common import logger
|
||||
from wrolpi.root_api import add_blueprint, json_response
|
||||
from wrolpi.schema import validate_doc, JSONErrorResponse
|
||||
from wrolpi.schema import JSONErrorResponse
|
||||
from .channel import lib as channel_lib
|
||||
from .channel.api import channel_bp
|
||||
from .lib import _refresh_videos, get_statistics
|
||||
|
|
@ -33,13 +35,9 @@ refresh_queue, refresh_event = create_websocket_feed('refresh', '/feeds/refresh'
|
|||
|
||||
@content_bp.post('/refresh')
|
||||
@content_bp.post('/refresh/<link:str>')
|
||||
@validate_doc(
|
||||
summary='Search for videos that have previously been downloaded and stored.',
|
||||
produces=StreamResponse,
|
||||
responses=[
|
||||
(HTTPStatus.BAD_REQUEST, JSONErrorResponse),
|
||||
],
|
||||
)
|
||||
@openapi.description('Search for videos that have previously been downloaded and stored.')
|
||||
@openapi.response(HTTPStatus.OK, StreamResponse)
|
||||
@openapi.response(HTTPStatus.BAD_REQUEST, JSONErrorResponse)
|
||||
@wrol_mode_check
|
||||
async def refresh(_, link: str = None):
|
||||
refresh_logger = logger.getChild('refresh')
|
||||
|
|
@ -75,13 +73,9 @@ download_queue, download_event = create_websocket_feed('download', '/feeds/downl
|
|||
|
||||
@content_bp.post('/download')
|
||||
@content_bp.post('/download/<link:str>')
|
||||
@validate_doc(
|
||||
summary='Update channel catalogs, download any missing videos',
|
||||
produces=StreamResponse,
|
||||
responses=[
|
||||
(HTTPStatus.BAD_REQUEST, JSONErrorResponse)
|
||||
],
|
||||
)
|
||||
@openapi.description('Update channel catalogs, download any missing videos')
|
||||
@openapi.response(HTTPStatus.OK, StreamResponse)
|
||||
@openapi.response(HTTPStatus.BAD_REQUEST, JSONErrorResponse)
|
||||
@wrol_mode_check
|
||||
def download(_, link: str = None):
|
||||
download_logger = logger.getChild('download')
|
||||
|
|
@ -105,14 +99,10 @@ async def refresh_videos(channel_links: list = None):
|
|||
|
||||
|
||||
@content_bp.post('/favorite')
|
||||
@validate_doc(
|
||||
summary='Toggle the favorite flag on a video',
|
||||
consumes=FavoriteRequest,
|
||||
produces=FavoriteResponse,
|
||||
responses=[
|
||||
(HTTPStatus.BAD_REQUEST, JSONErrorResponse)
|
||||
]
|
||||
)
|
||||
@openapi.description('Toggle the favorite flag on a video')
|
||||
@validate(FavoriteRequest)
|
||||
@openapi.response(HTTPStatus.OK, FavoriteResponse)
|
||||
@openapi.response(HTTPStatus.BAD_REQUEST, JSONErrorResponse)
|
||||
async def favorite(_: Request, data: dict):
|
||||
_favorite = video_lib.set_video_favorite(data['video_id'], data['favorite'])
|
||||
ret = {'video_id': data['video_id'], 'favorite': _favorite}
|
||||
|
|
@ -120,10 +110,7 @@ async def favorite(_: Request, data: dict):
|
|||
|
||||
|
||||
@content_bp.get('/statistics')
|
||||
@validate_doc(
|
||||
summary='Retrieve video statistics',
|
||||
produces=VideosStatisticsResponse,
|
||||
)
|
||||
@openapi.response(HTTPStatus.OK, VideosStatisticsResponse)
|
||||
async def statistics(_: Request):
|
||||
ret = await get_statistics()
|
||||
return json_response(ret, HTTPStatus.OK)
|
||||
|
|
|
|||
|
|
@ -3,11 +3,13 @@ from http import HTTPStatus
|
|||
|
||||
from sanic import response, Blueprint
|
||||
from sanic.request import Request
|
||||
from sanic_ext import validate
|
||||
from sanic_ext.extensions.openapi import openapi
|
||||
|
||||
from wrolpi.common import logger, wrol_mode_check, make_media_directory, \
|
||||
get_media_directory
|
||||
from wrolpi.root_api import json_response
|
||||
from wrolpi.schema import validate_doc, JSONErrorResponse
|
||||
from wrolpi.schema import JSONErrorResponse
|
||||
from .lib import get_minimal_channels, delete_channel, update_channel, get_channel, create_channel
|
||||
from ..schema import ChannelsResponse, ChannelResponse, ChannelPostRequest, \
|
||||
ChannelPostResponse, ChannelPutRequest, SuccessResponse
|
||||
|
|
@ -18,23 +20,16 @@ logger = logger.getChild(__name__)
|
|||
|
||||
|
||||
@channel_bp.get('/')
|
||||
@validate_doc(
|
||||
summary='Get a list of all Channels',
|
||||
produces=ChannelsResponse,
|
||||
)
|
||||
@openapi.description('Get a list of all Channels')
|
||||
@openapi.response(HTTPStatus.OK, ChannelsResponse)
|
||||
async def get_channels(_: Request):
|
||||
channels = await get_minimal_channels()
|
||||
return json_response({'channels': channels})
|
||||
|
||||
|
||||
@channel_bp.route('/<link:str>', methods=['GET', 'OPTIONS'])
|
||||
@validate_doc(
|
||||
summary='Get a Channel',
|
||||
produces=ChannelResponse,
|
||||
responses=(
|
||||
(HTTPStatus.NOT_FOUND, JSONErrorResponse),
|
||||
),
|
||||
)
|
||||
@openapi.description('Get a Channel')
|
||||
@openapi.response(HTTPStatus.OK, ChannelResponse)
|
||||
def channel_get(_: Request, link: str):
|
||||
channel = get_channel(link=link)
|
||||
channel.pop('info_json')
|
||||
|
|
@ -42,16 +37,21 @@ def channel_get(_: Request, link: str):
|
|||
|
||||
|
||||
@channel_bp.post('/')
|
||||
@validate_doc(
|
||||
summary='Insert a Channel',
|
||||
consumes=ChannelPostRequest,
|
||||
responses=(
|
||||
(HTTPStatus.CREATED, ChannelPostResponse),
|
||||
(HTTPStatus.BAD_REQUEST, JSONErrorResponse),
|
||||
),
|
||||
)
|
||||
@openapi.description('Insert a Channel')
|
||||
@openapi.response(HTTPStatus.OK, ChannelPostResponse)
|
||||
@openapi.response(HTTPStatus.BAD_REQUEST, JSONErrorResponse)
|
||||
@validate(ChannelPostRequest)
|
||||
@wrol_mode_check
|
||||
def channel_post(_: Request, data: dict):
|
||||
try:
|
||||
# Channel directory is relative to the media directory. Channel directory may not be in "videos" directory!
|
||||
data['directory'] = get_media_directory() / data['directory']
|
||||
except UnknownDirectory:
|
||||
if data['mkdir']:
|
||||
make_media_directory(data['directory'])
|
||||
data['directory'] = get_relative_to_media_directory(data['directory'])
|
||||
else:
|
||||
raise
|
||||
# Channel directory is relative to the media directory. Channel directory may not be in "videos" directory!
|
||||
data['directory'] = get_media_directory() / data['directory']
|
||||
if data.get('mkdir') is True:
|
||||
|
|
@ -79,14 +79,10 @@ def channel_post(_: Request, data: dict):
|
|||
|
||||
@channel_bp.put('/<link:str>')
|
||||
@channel_bp.patch('/<link:str>')
|
||||
@validate_doc(
|
||||
summary='Update a Channel',
|
||||
consumes=ChannelPutRequest,
|
||||
responses=(
|
||||
(HTTPStatus.NOT_FOUND, JSONErrorResponse),
|
||||
(HTTPStatus.BAD_REQUEST, JSONErrorResponse),
|
||||
),
|
||||
)
|
||||
@openapi.description('Update a Channel')
|
||||
@openapi.response(HTTPStatus.NO_CONTENT)
|
||||
@openapi.response(HTTPStatus.BAD_REQUEST, JSONErrorResponse)
|
||||
@validate(ChannelPutRequest)
|
||||
@wrol_mode_check
|
||||
def channel_update(_: Request, link: str, data: dict):
|
||||
channel = update_channel(data, link)
|
||||
|
|
@ -95,13 +91,9 @@ def channel_update(_: Request, link: str, data: dict):
|
|||
|
||||
|
||||
@channel_bp.delete('/<link:str>')
|
||||
@validate_doc(
|
||||
summary='Delete a Channel',
|
||||
produces=SuccessResponse,
|
||||
responses=(
|
||||
(HTTPStatus.NOT_FOUND, JSONErrorResponse),
|
||||
),
|
||||
)
|
||||
@openapi.description('Delete a Channel')
|
||||
@openapi.response(HTTPStatus.NOT_FOUND, JSONErrorResponse)
|
||||
@validate(SuccessResponse)
|
||||
@wrol_mode_check
|
||||
def channel_delete(_: Request, link: str):
|
||||
delete_channel(link)
|
||||
|
|
|
|||
|
|
@ -1,179 +1,209 @@
|
|||
from sanic_openapi import doc
|
||||
from dataclasses import dataclass, field
|
||||
from datetime import date, datetime
|
||||
from typing import List, Optional
|
||||
|
||||
|
||||
@dataclass
|
||||
class ChannelPostRequest:
|
||||
url = doc.String()
|
||||
name = doc.String(description='A short readable name', required=True)
|
||||
match_regex = doc.String(description='Regex that the video title will have to match')
|
||||
directory = doc.String(required=True)
|
||||
mkdir = doc.Boolean()
|
||||
generate_posters = doc.Boolean()
|
||||
calculate_duration = doc.Boolean()
|
||||
download_frequency = doc.String()
|
||||
url: str
|
||||
name: str
|
||||
match_regex: str
|
||||
directory: str
|
||||
mkdir: bool
|
||||
generate_posters: bool
|
||||
calculate_duration: bool
|
||||
download_frequency: str
|
||||
|
||||
|
||||
@dataclass
|
||||
class ChannelPutRequest:
|
||||
url = doc.String()
|
||||
name = doc.String()
|
||||
match_regex = doc.String()
|
||||
link = doc.String()
|
||||
directory = doc.String()
|
||||
generate_posters = doc.Boolean()
|
||||
calculate_duration = doc.Boolean()
|
||||
mkdir = doc.Boolean()
|
||||
download_frequency = doc.String()
|
||||
url: str
|
||||
name: str
|
||||
match_regex: str
|
||||
link: str
|
||||
directory: str
|
||||
generate_posters: bool
|
||||
calculate_duration: bool
|
||||
mkdir: bool
|
||||
download_frequency: str
|
||||
|
||||
|
||||
@dataclass
|
||||
class RefreshRequest:
|
||||
channel_name = doc.String()
|
||||
channel_name: str
|
||||
|
||||
|
||||
@dataclass
|
||||
class SuccessResponse:
|
||||
success = doc.String()
|
||||
success: str
|
||||
|
||||
|
||||
@dataclass
|
||||
class ChannelModel:
|
||||
id = doc.Integer()
|
||||
url = doc.String()
|
||||
name = doc.String()
|
||||
match_regex = doc.String()
|
||||
link = doc.String()
|
||||
directory = doc.String()
|
||||
id: int
|
||||
url: str
|
||||
name: str
|
||||
match_regex: str
|
||||
link: str
|
||||
directory: str
|
||||
|
||||
|
||||
@dataclass
|
||||
class ChannelResponse:
|
||||
channel = ChannelModel
|
||||
|
||||
|
||||
@dataclass
|
||||
class ChannelPostResponse:
|
||||
success = doc.String()
|
||||
success: str
|
||||
|
||||
|
||||
@dataclass
|
||||
class ChannelsChannelModel:
|
||||
id = doc.Integer()
|
||||
name = doc.String()
|
||||
link = doc.String()
|
||||
id: int
|
||||
name: str
|
||||
link: str
|
||||
|
||||
|
||||
@dataclass
|
||||
class ChannelsResponse:
|
||||
channels = doc.List(doc.Object(ChannelsChannelModel))
|
||||
channels: List[ChannelsChannelModel]
|
||||
|
||||
|
||||
@dataclass
|
||||
class StreamResponse:
|
||||
success = doc.String()
|
||||
stream_url = doc.String()
|
||||
success: str
|
||||
stream_url: str
|
||||
|
||||
|
||||
@dataclass
|
||||
class Video:
|
||||
id = doc.Integer()
|
||||
description_path = doc.String()
|
||||
ext = doc.String()
|
||||
poster_path = doc.String()
|
||||
source_id = doc.String()
|
||||
title = doc.String()
|
||||
upload_date = doc.Date()
|
||||
video_path = doc.String()
|
||||
name = doc.String()
|
||||
caption_path = doc.String()
|
||||
idempotency = doc.String()
|
||||
info_json_path = doc.String()
|
||||
channel_id = doc.Integer()
|
||||
id: int
|
||||
description_path: str
|
||||
ext: str
|
||||
poster_path: str
|
||||
source_id: str
|
||||
title: str
|
||||
upload_date: date
|
||||
video_path: str
|
||||
name: str
|
||||
caption_path: str
|
||||
idempotency: str
|
||||
info_json_path: str
|
||||
channel_id: int
|
||||
|
||||
|
||||
@dataclass
|
||||
class VideoWithChannel:
|
||||
id = doc.Integer()
|
||||
description_path = doc.String()
|
||||
ext = doc.String()
|
||||
poster_path = doc.String()
|
||||
source_id = doc.String()
|
||||
title = doc.String()
|
||||
upload_date = doc.Date()
|
||||
video_path = doc.String()
|
||||
name = doc.String()
|
||||
caption_path = doc.String()
|
||||
idempotency = doc.String()
|
||||
info_json_path = doc.String()
|
||||
channel_id = doc.Integer()
|
||||
id: int
|
||||
description_path: str
|
||||
ext: str
|
||||
poster_path: str
|
||||
source_id: str
|
||||
title: str
|
||||
upload_date: date
|
||||
video_path: str
|
||||
name: str
|
||||
caption_path: str
|
||||
idempotency: str
|
||||
info_json_path: str
|
||||
channel_id: int
|
||||
|
||||
channel = ChannelModel
|
||||
|
||||
|
||||
@dataclass
|
||||
class ChannelVideosResponse:
|
||||
videos = doc.List(Video)
|
||||
videos: List[Video]
|
||||
|
||||
|
||||
@dataclass
|
||||
class ExtendedVideo(Video):
|
||||
info_json = doc.Object(dict)
|
||||
info_json: dict
|
||||
|
||||
|
||||
@dataclass
|
||||
class VideoResponse:
|
||||
video = ExtendedVideo
|
||||
|
||||
|
||||
@dataclass
|
||||
class VideoSearchRequest:
|
||||
search_str = doc.String()
|
||||
channel_link = doc.String()
|
||||
order_by = doc.String()
|
||||
offset = doc.Integer()
|
||||
limit = doc.Integer()
|
||||
filters = doc.List(doc.String())
|
||||
search_str: str
|
||||
filters: Optional[List[str]] = field(default_factory=list)
|
||||
offset: Optional[int] = None
|
||||
limit: Optional[int] = None
|
||||
order_by: Optional[str] = None
|
||||
channel_link: Optional[str] = None
|
||||
|
||||
|
||||
@dataclass
|
||||
class VideoSearchResponse:
|
||||
videos = doc.List(VideoWithChannel)
|
||||
tsquery = doc.String()
|
||||
videos: List[VideoWithChannel]
|
||||
tsquery: str
|
||||
|
||||
|
||||
@dataclass
|
||||
class PaginationQuery:
|
||||
offset = doc.Integer()
|
||||
offset: int
|
||||
|
||||
|
||||
@dataclass
|
||||
class DirectoriesRequest:
|
||||
search_str = doc.String(required=True)
|
||||
search_str: str
|
||||
|
||||
|
||||
@dataclass
|
||||
class DirectoriesResponse:
|
||||
directories = doc.List(doc.String())
|
||||
directories: List[str]
|
||||
|
||||
|
||||
@dataclass
|
||||
class FavoriteRequest:
|
||||
video_id = doc.Integer()
|
||||
favorite = doc.Boolean()
|
||||
video_id: int
|
||||
favorite: bool
|
||||
|
||||
|
||||
@dataclass
|
||||
class FavoriteResponse:
|
||||
video_id = doc.Integer()
|
||||
favorite = doc.DateTime()
|
||||
video_id: int
|
||||
favorite: datetime
|
||||
|
||||
|
||||
@dataclass
|
||||
class WROLModeRequest:
|
||||
enabled = doc.Boolean(required=True)
|
||||
enabled: bool
|
||||
|
||||
|
||||
@dataclass
|
||||
class VideoStatistics:
|
||||
videos = doc.Integer()
|
||||
favorites = doc.Integer()
|
||||
week = doc.Integer()
|
||||
month = doc.Integer()
|
||||
year = doc.Integer()
|
||||
sum_duration = doc.Integer()
|
||||
sum_size = doc.Integer()
|
||||
max_size = doc.Integer()
|
||||
videos: int
|
||||
favorites: int
|
||||
week: int
|
||||
month: int
|
||||
year: int
|
||||
sum_duration: int
|
||||
sum_size: int
|
||||
max_size: int
|
||||
|
||||
|
||||
@dataclass
|
||||
class ChannelStatistics:
|
||||
channels = doc.Integer()
|
||||
channels: int
|
||||
|
||||
|
||||
@dataclass
|
||||
class VideosStatisticsResponse:
|
||||
videos = VideoStatistics
|
||||
channels = ChannelStatistics
|
||||
|
||||
|
||||
@dataclass
|
||||
class CensoredVideoRequest:
|
||||
limit = doc.Integer()
|
||||
offset = doc.Integer()
|
||||
limit: int
|
||||
offset: int
|
||||
|
||||
|
||||
@dataclass
|
||||
class CensoredVideoResponse:
|
||||
videos = doc.List(VideoWithChannel)
|
||||
videos: List[VideoWithChannel]
|
||||
|
|
|
|||
|
|
@ -2,13 +2,15 @@ from http import HTTPStatus
|
|||
|
||||
from sanic import response, Blueprint
|
||||
from sanic.request import Request
|
||||
from sanic_ext import validate
|
||||
from sanic_ext.extensions.openapi import openapi
|
||||
|
||||
from wrolpi.common import logger, wrol_mode_check, run_after, get_media_directory, \
|
||||
get_relative_to_media_directory
|
||||
from wrolpi.db import get_db_session
|
||||
from wrolpi.errors import ValidationError, InvalidOrderBy
|
||||
from wrolpi.root_api import json_response
|
||||
from wrolpi.schema import validate_doc, JSONErrorResponse
|
||||
from wrolpi.schema import JSONErrorResponse
|
||||
from .lib import get_video, VIDEO_ORDERS, DEFAULT_VIDEO_ORDER, video_search, get_video_for_app
|
||||
from ..common import get_matching_directories, video_limiter
|
||||
from ..lib import save_channels_config
|
||||
|
|
@ -21,24 +23,19 @@ logger = logger.getChild(__name__)
|
|||
|
||||
|
||||
@video_bp.get('/video/<video_id:int>')
|
||||
@validate_doc(
|
||||
summary='Get Video information',
|
||||
produces=VideoResponse,
|
||||
responses=(
|
||||
(HTTPStatus.NOT_FOUND, JSONErrorResponse),
|
||||
),
|
||||
)
|
||||
@openapi.description('Get Video information')
|
||||
@openapi.response(HTTPStatus.OK, VideoResponse)
|
||||
@openapi.response(HTTPStatus.NOT_FOUND, JSONErrorResponse)
|
||||
def video_get(_: Request, video_id: int):
|
||||
video, previous_video, next_video = get_video_for_app(video_id)
|
||||
return json_response({'video': video, 'prev': previous_video, 'next': next_video})
|
||||
|
||||
|
||||
@video_bp.post('/search')
|
||||
@validate_doc(
|
||||
summary='Search Video titles and captions',
|
||||
consumes=VideoSearchRequest,
|
||||
produces=VideoSearchResponse,
|
||||
)
|
||||
@openapi.description('Search Video titles and captions')
|
||||
@openapi.response(HTTPStatus.OK, VideoSearchResponse)
|
||||
@openapi.response(HTTPStatus.NOT_FOUND, JSONErrorResponse)
|
||||
@validate(VideoSearchRequest)
|
||||
async def search(_: Request, data: dict):
|
||||
try:
|
||||
search_str = data.get('search_str')
|
||||
|
|
@ -60,14 +57,10 @@ async def search(_: Request, data: dict):
|
|||
|
||||
|
||||
@video_bp.post('/directories')
|
||||
@validate_doc(
|
||||
summary='Get all directories that match the search_str, prefixed by the media directory.',
|
||||
consumes=DirectoriesRequest,
|
||||
responses=(
|
||||
(HTTPStatus.NOT_FOUND, JSONErrorResponse),
|
||||
(HTTPStatus.OK, DirectoriesResponse),
|
||||
),
|
||||
)
|
||||
@openapi.description('Get all directories that match the search_str, prefixed by the media directory.')
|
||||
@openapi.response(HTTPStatus.OK, DirectoriesResponse)
|
||||
@openapi.response(HTTPStatus.NOT_FOUND, JSONErrorResponse)
|
||||
@validate(DirectoriesRequest)
|
||||
def directories(_, data):
|
||||
search_str = str(get_media_directory() / data['search_str'])
|
||||
dirs = get_matching_directories(search_str)
|
||||
|
|
@ -76,12 +69,9 @@ def directories(_, data):
|
|||
|
||||
|
||||
@video_bp.delete('/video/<video_id:int>')
|
||||
@validate_doc(
|
||||
summary='Delete a video',
|
||||
responses=(
|
||||
(HTTPStatus.NOT_FOUND, JSONErrorResponse),
|
||||
),
|
||||
)
|
||||
@openapi.description('Delete a video.')
|
||||
@openapi.response(HTTPStatus.NO_CONTENT)
|
||||
@openapi.response(HTTPStatus.NOT_FOUND, JSONErrorResponse)
|
||||
@wrol_mode_check
|
||||
@run_after(save_channels_config)
|
||||
def video_delete(_: Request, video_id: int):
|
||||
|
|
|
|||
|
|
@ -10,10 +10,9 @@ pytest==6.2.5
|
|||
pytz==2021.3
|
||||
PyYAML==5.3.1
|
||||
requests==2.27.1
|
||||
Sanic-Cors==1.0.1
|
||||
sanic-openapi==0.6.2
|
||||
sanic-testing==0.7.0
|
||||
sanic==21.9.1
|
||||
sanic-ext==21.12.1
|
||||
sanic-testing==0.8.2
|
||||
sanic==21.12.1
|
||||
selenium==4.1.0
|
||||
SQLAlchemy==1.3.22
|
||||
srt==3.5.0
|
||||
|
|
|
|||
|
|
@ -11,7 +11,8 @@ from pytz import UnknownTimeZoneError
|
|||
from sanic import Sanic, response, Blueprint, __version__ as sanic_version
|
||||
from sanic.request import Request
|
||||
from sanic.response import HTTPResponse
|
||||
from sanic_cors import CORS
|
||||
from sanic_ext import validate
|
||||
from sanic_ext.extensions.openapi import openapi
|
||||
|
||||
from wrolpi.common import set_sanic_url_parts, logger, get_config, wrol_mode_enabled, save_settings_config, \
|
||||
Base, get_media_directory, wrol_mode_check
|
||||
|
|
@ -19,8 +20,7 @@ from wrolpi.dates import set_timezone
|
|||
from wrolpi.downloader import download_manager
|
||||
from wrolpi.errors import WROLModeEnabled, InvalidTimezone
|
||||
from wrolpi.media_path import MediaPath
|
||||
from wrolpi.schema import RegexRequest, RegexResponse, SettingsRequest, SettingsResponse, EchoResponse, \
|
||||
validate_doc, DownloadRequest
|
||||
from wrolpi.schema import RegexRequest, RegexResponse, SettingsRequest, SettingsResponse, DownloadRequest, EchoResponse
|
||||
from wrolpi.vars import DOCKERIZED
|
||||
from wrolpi.version import __version__
|
||||
|
||||
|
|
@ -30,17 +30,6 @@ api_app = Sanic(name='api_app')
|
|||
|
||||
DEFAULT_HOST, DEFAULT_PORT = '127.0.0.1', '8081'
|
||||
|
||||
# TODO Allow all requests to this webapp during development. This should be restricted later.
|
||||
CORS(
|
||||
api_app,
|
||||
expose_headers=[
|
||||
'Location', # Expose this header so the App can send users to the location of a created object.
|
||||
],
|
||||
resources={
|
||||
'/*': {'origins': '*'},
|
||||
}
|
||||
)
|
||||
|
||||
root_api = Blueprint('RootAPI', url_prefix='/api')
|
||||
|
||||
BLUEPRINTS = [root_api, ]
|
||||
|
|
@ -105,11 +94,8 @@ def index(_):
|
|||
|
||||
|
||||
@root_api.route('/echo', methods=['GET', 'POST', 'PUT', 'DELETE', 'PATCH', 'OPTIONS'])
|
||||
@validate_doc(
|
||||
summary='Echo whatever is sent to this',
|
||||
produces=EchoResponse,
|
||||
tag='Testing',
|
||||
)
|
||||
@openapi.description('Echo whatever is sent to this.')
|
||||
@openapi.response(HTTPStatus.OK, EchoResponse)
|
||||
async def echo(request: Request):
|
||||
"""
|
||||
Returns a JSON object containing details about the request sent to me.
|
||||
|
|
@ -125,20 +111,16 @@ async def echo(request: Request):
|
|||
|
||||
|
||||
@root_api.route('/settings', methods=['GET', 'OPTIONS'])
|
||||
@validate_doc(
|
||||
summary='Get WROLPi settings',
|
||||
produces=SettingsResponse,
|
||||
)
|
||||
@openapi.description('Get WROLPi settings')
|
||||
@openapi.response(HTTPStatus.OK, SettingsResponse)
|
||||
def get_settings(_: Request):
|
||||
config = dict(get_config())
|
||||
return json_response({'config': config, 'version': __version__})
|
||||
|
||||
|
||||
@root_api.patch('/settings')
|
||||
@validate_doc(
|
||||
summary='Update WROLPi settings',
|
||||
consumes=SettingsRequest,
|
||||
)
|
||||
@openapi.description('Update WROLPi settings')
|
||||
@validate(json=SettingsRequest)
|
||||
def update_settings(_: Request, data: dict):
|
||||
if wrol_mode_enabled() and 'wrol_mode' not in data:
|
||||
# Cannot update settings while WROL Mode is enabled, unless you want to disable WROL Mode.
|
||||
|
|
@ -159,32 +141,26 @@ def update_settings(_: Request, data: dict):
|
|||
|
||||
|
||||
@root_api.post('/valid_regex')
|
||||
@validate_doc(
|
||||
summary='Check if the regex is valid',
|
||||
consumes=RegexRequest,
|
||||
responses=(
|
||||
(HTTPStatus.OK, RegexResponse),
|
||||
(HTTPStatus.BAD_REQUEST, RegexResponse),
|
||||
)
|
||||
)
|
||||
def valid_regex(_: Request, data: dict):
|
||||
@openapi.description('Check if the regex is valid.')
|
||||
@openapi.response(HTTPStatus.OK, RegexResponse)
|
||||
@openapi.response(HTTPStatus.BAD_REQUEST, RegexResponse)
|
||||
@validate(RegexRequest)
|
||||
def valid_regex(_: Request, body: RegexRequest):
|
||||
try:
|
||||
re.compile(data['regex'])
|
||||
return response.json({'valid': True, 'regex': data['regex']})
|
||||
re.compile(body.regex)
|
||||
return response.json({'valid': True, 'regex': body.regex})
|
||||
except re.error:
|
||||
return response.json({'valid': False, 'regex': data['regex']}, HTTPStatus.BAD_REQUEST)
|
||||
return response.json({'valid': False, 'regex': body.regex}, HTTPStatus.BAD_REQUEST)
|
||||
|
||||
|
||||
@root_api.post('/download')
|
||||
@validate_doc(
|
||||
summary='Download the many URLs that are provided.',
|
||||
consumes=DownloadRequest,
|
||||
)
|
||||
@openapi.description('Download the many URLs that are provided.')
|
||||
@validate(DownloadRequest)
|
||||
@wrol_mode_check
|
||||
async def post_download(request: Request, data: dict):
|
||||
async def post_download(_: Request, body: DownloadRequest):
|
||||
# URLs are provided in a textarea, lets split all lines.
|
||||
urls = [i.strip() for i in str(data['urls']).strip().splitlines()]
|
||||
downloader = data.get('downloader')
|
||||
urls = [i.strip() for i in str(body.urls).strip().splitlines()]
|
||||
downloader = body.downloader
|
||||
if not downloader or downloader in ('auto', 'None', 'null'):
|
||||
downloader = None
|
||||
download_manager.create_downloads(urls, downloader=downloader, reset_attempts=True)
|
||||
|
|
@ -192,41 +168,44 @@ async def post_download(request: Request, data: dict):
|
|||
|
||||
|
||||
@root_api.get('/download')
|
||||
@validate_doc(
|
||||
summary='Get all Downloads that need to be processed.',
|
||||
)
|
||||
async def get_downloads(request: Request):
|
||||
@openapi.description('Get all Downloads that need to be processed.')
|
||||
async def get_downloads(_: Request):
|
||||
data = download_manager.get_fe_downloads()
|
||||
return json_response(data)
|
||||
|
||||
|
||||
@root_api.post('/download/<download_id:int>/kill')
|
||||
async def kill_download(request: Request, download_id: int):
|
||||
@openapi.description('Kill a download. It will be stopped if it is pending.')
|
||||
async def kill_download(_: Request, download_id: int):
|
||||
download_manager.kill_download(download_id)
|
||||
return response.empty()
|
||||
|
||||
|
||||
@root_api.post('/download/kill')
|
||||
async def kill_downloads(request: Request):
|
||||
@openapi.description('Kill all downloads. Disable downloading.')
|
||||
async def kill_downloads(_: Request):
|
||||
download_manager.kill()
|
||||
return response.empty()
|
||||
|
||||
|
||||
@root_api.post('/download/enable')
|
||||
async def enable_downloads(request: Request):
|
||||
@openapi.description('Enable and start downloading.')
|
||||
async def enable_downloads(_: Request):
|
||||
download_manager.enable()
|
||||
return response.empty()
|
||||
|
||||
|
||||
@root_api.delete('/download/<download_id:integer>')
|
||||
@openapi.description('Delete a download')
|
||||
@wrol_mode_check
|
||||
async def delete_download(request: Request, download_id: int):
|
||||
async def delete_download(_: Request, download_id: int):
|
||||
deleted = download_manager.delete_download(download_id)
|
||||
return response.empty(HTTPStatus.NO_CONTENT if deleted else HTTPStatus.NOT_FOUND)
|
||||
|
||||
|
||||
@root_api.get('/downloaders')
|
||||
async def get_downloaders(request: Request):
|
||||
@openapi.description('List all Downloaders that can be specified by the user.')
|
||||
async def get_downloaders(_: Request):
|
||||
downloaders = download_manager.list_downloaders()
|
||||
downloaders.insert(0, dict(name='auto', pretty_name='Automatic'))
|
||||
disabled = download_manager.disabled.is_set()
|
||||
|
|
|
|||
207
wrolpi/schema.py
207
wrolpi/schema.py
|
|
@ -1,207 +1,60 @@
|
|||
import inspect
|
||||
from datetime import datetime
|
||||
from functools import wraps
|
||||
from uuid import UUID
|
||||
|
||||
import sanic
|
||||
from sanic import response
|
||||
from sanic_openapi import doc
|
||||
from sanic_openapi.doc import Field
|
||||
|
||||
from wrolpi.common import string_to_boolean, logger
|
||||
from wrolpi.errors import NoBodyContents, ValidationError, MissingRequiredField, ExcessJSONFields, API_ERRORS, APIError
|
||||
from wrolpi.vars import DATE_FORMAT
|
||||
|
||||
|
||||
def validate_data(model: type, data: dict, optional_body: bool = False):
|
||||
"""
|
||||
Convert a JSON object to the model's specification. If the JSON object matches the model's specification, this
|
||||
function will return a python dict of that data. If it doesn't match, this will return a Sanic response object
|
||||
containing an error.
|
||||
|
||||
If `optional_body` is True, then the request body may be empty. Otherwise, raise NoBodyContents.
|
||||
"""
|
||||
if optional_body and not data:
|
||||
# Body was empty in the request, but body is not required.
|
||||
return {}
|
||||
|
||||
if not data:
|
||||
raise NoBodyContents()
|
||||
|
||||
new_data = {}
|
||||
# Get the public attributes of the model
|
||||
attrs = [i for i in dir(model) if not str(i).startswith('__')]
|
||||
# Convert each json value to it's respective doc field's python type
|
||||
# i.e. "json value" -> doc.String -> str
|
||||
missing_fields = []
|
||||
for attr in attrs:
|
||||
field = getattr(model, attr)
|
||||
try:
|
||||
if isinstance(field, doc.String):
|
||||
new_data[attr] = str(data.pop(attr))
|
||||
elif isinstance(field, doc.Integer):
|
||||
new_data[attr] = int(data.pop(attr))
|
||||
elif isinstance(field, doc.Tuple):
|
||||
new_data[attr] = tuple(data.pop(attr))
|
||||
elif isinstance(field, doc.UUID):
|
||||
new_data[attr] = UUID(data.pop(attr))
|
||||
elif isinstance(field, doc.Boolean):
|
||||
new_data[attr] = string_to_boolean(data.pop(attr))
|
||||
elif isinstance(field, doc.Float):
|
||||
val = data.pop(attr)
|
||||
new_data[attr] = float(val) if val else None
|
||||
elif isinstance(field, doc.Dictionary):
|
||||
new_data[attr] = dict(data.pop(attr))
|
||||
elif isinstance(field, doc.List):
|
||||
new_data[attr] = list(data.pop(attr))
|
||||
elif isinstance(field, Trinary):
|
||||
val = data.pop(attr)
|
||||
if val is not None:
|
||||
val = string_to_boolean(val)
|
||||
new_data[attr] = val
|
||||
elif isinstance(field, doc.Date):
|
||||
val = data.pop(attr)
|
||||
if val:
|
||||
new_data[attr] = datetime.strptime(val, DATE_FORMAT)
|
||||
elif isinstance(field, doc.DateTime):
|
||||
val = data.pop(attr)
|
||||
if val:
|
||||
new_data[attr] = datetime.utcfromtimestamp(val)
|
||||
else:
|
||||
raise ValidationError(f'Bad field type {field} specified in the API model!')
|
||||
except KeyError:
|
||||
if field.required:
|
||||
missing_fields.append(attr)
|
||||
|
||||
if missing_fields:
|
||||
raise MissingRequiredField(f'Missing fields: {missing_fields}')
|
||||
|
||||
if data:
|
||||
raise ExcessJSONFields(f'Extra fields: {data.keys()}')
|
||||
|
||||
return new_data
|
||||
|
||||
|
||||
def validate_doc(summary: str = None, consumes=None, produces=None, responses=(), tag: str = None,
|
||||
optional_body: bool = False):
|
||||
"""
|
||||
Apply Sanic OpenAPI docs to the wrapped route. Perform simple validation on requests.
|
||||
"""
|
||||
|
||||
def wrapper(func):
|
||||
@wraps(func)
|
||||
async def wrapped(request, *a, **kw):
|
||||
try:
|
||||
if consumes:
|
||||
if 'data' in kw:
|
||||
raise OverflowError(f'data kwarg already being passed to {func}')
|
||||
|
||||
data = validate_data(consumes, request.json, optional_body)
|
||||
if isinstance(data, sanic.response.HTTPResponse):
|
||||
# Error in validation
|
||||
return data
|
||||
kw['data'] = data
|
||||
|
||||
result = func(request, *a, **kw)
|
||||
if inspect.iscoroutine(result):
|
||||
result = await result
|
||||
return result
|
||||
except ValidationError as e:
|
||||
error = API_ERRORS[type(e)]
|
||||
body = {
|
||||
'error': error['message'],
|
||||
'code': error['code'],
|
||||
}
|
||||
logger.error(e, exc_info=True)
|
||||
|
||||
if e.__cause__:
|
||||
cause = e.__cause__
|
||||
cause = API_ERRORS[type(cause)] if cause else None
|
||||
body['cause'] = {'error': cause['message'], 'code': cause['code']}
|
||||
|
||||
r = response.json(body, error['status'])
|
||||
return r
|
||||
except APIError as e:
|
||||
# The endpoint returned a standardized APIError, convert it to a json response
|
||||
error = API_ERRORS[type(e)]
|
||||
r = response.json({'message': str(e), 'api_error': error['message'], 'code': error['code']},
|
||||
error['status'])
|
||||
return r
|
||||
|
||||
# Apply the docs to the wrapped function so sanic-openapi can find the wrapped function when
|
||||
# building the schema. If these docs are applied to `func`, sanic-openapi won't be able to lookup `wrapped`
|
||||
if summary:
|
||||
wrapped = doc.summary(summary)(wrapped)
|
||||
if consumes:
|
||||
wrapped = doc.consumes(consumes, location='body')(wrapped)
|
||||
if produces:
|
||||
wrapped = doc.produces(produces)(wrapped)
|
||||
for resp in responses:
|
||||
wrapped = doc.response(*resp)(wrapped)
|
||||
if tag:
|
||||
wrapped = doc.tag(tag)(wrapped)
|
||||
|
||||
return wrapped
|
||||
|
||||
return wrapper
|
||||
from dataclasses import dataclass
|
||||
from typing import List
|
||||
|
||||
|
||||
@dataclass
|
||||
class SettingsObject:
|
||||
media_directory = doc.String()
|
||||
media_directory: str
|
||||
|
||||
|
||||
@dataclass
|
||||
class SettingsResponse:
|
||||
config = doc.Object(SettingsObject)
|
||||
config: SettingsObject
|
||||
|
||||
|
||||
@dataclass
|
||||
class SettingsRequest:
|
||||
media_directory = doc.String()
|
||||
wrol_mode = doc.Boolean()
|
||||
timezone = doc.String()
|
||||
media_directory: str
|
||||
wrol_mode: bool
|
||||
timezone: str
|
||||
|
||||
|
||||
@dataclass
|
||||
class RegexRequest:
|
||||
regex = doc.String()
|
||||
regex: str
|
||||
|
||||
|
||||
@dataclass
|
||||
class RegexResponse:
|
||||
regex = doc.String()
|
||||
valid = doc.Boolean()
|
||||
regex: str
|
||||
valid: bool
|
||||
|
||||
|
||||
@dataclass
|
||||
class EchoResponse:
|
||||
form = doc.Dictionary()
|
||||
headers = doc.Dictionary()
|
||||
json = doc.String()
|
||||
method = doc.String()
|
||||
form: dict
|
||||
headers: dict
|
||||
json: str
|
||||
method: str
|
||||
|
||||
|
||||
@dataclass
|
||||
class EventObject:
|
||||
name = doc.String()
|
||||
is_set = doc.Boolean()
|
||||
name: str
|
||||
is_set: str
|
||||
|
||||
|
||||
@dataclass
|
||||
class EventsResponse:
|
||||
events = doc.List(EventObject)
|
||||
events: List[EventObject]
|
||||
|
||||
|
||||
@dataclass
|
||||
class DownloadRequest:
|
||||
urls = doc.String(required=True)
|
||||
downloader = doc.String()
|
||||
|
||||
|
||||
class Trinary(Field):
|
||||
"""
|
||||
A field for API docs. Can be True/False/None.
|
||||
"""
|
||||
|
||||
def __init__(self, *a, **kw):
|
||||
kw['choices'] = (True, False, None)
|
||||
super().__init__(*a, **kw)
|
||||
|
||||
def serialize(self):
|
||||
return {"type": "trinary", **super().serialize()}
|
||||
urls: str
|
||||
downloader: str
|
||||
|
||||
|
||||
@dataclass
|
||||
class JSONErrorResponse:
|
||||
error = doc.String()
|
||||
error: str
|
||||
|
|
|
|||
|
|
@ -12,7 +12,7 @@ from typing import List, Optional
|
|||
import mock
|
||||
import websockets
|
||||
import yaml
|
||||
from sanic_openapi.api import Response
|
||||
from requests import Response
|
||||
|
||||
from wrolpi.common import EXAMPLE_CONFIG_PATH, get_config, set_test_media_directory, get_media_directory
|
||||
from wrolpi.conftest import ROUTES_ATTACHED, test_db, test_client # noqa
|
||||
|
|
|
|||
|
|
@ -6,76 +6,13 @@ from datetime import date, datetime
|
|||
from decimal import Decimal
|
||||
|
||||
import pytest
|
||||
from sanic_openapi import doc
|
||||
|
||||
from wrolpi.common import combine_dicts, insert_parameter, date_range, api_param_limiter, chdir, zig_zag
|
||||
from wrolpi.dates import set_timezone, now
|
||||
from wrolpi.errors import NoBodyContents, MissingRequiredField, ExcessJSONFields, InvalidTimezone
|
||||
from wrolpi.schema import validate_data
|
||||
from wrolpi.errors import InvalidTimezone
|
||||
from wrolpi.test.common import build_test_directories, wrap_media_directory
|
||||
|
||||
|
||||
class Model:
|
||||
name = doc.String(required=True)
|
||||
count = doc.Integer(required=True)
|
||||
fish = doc.Tuple()
|
||||
on = doc.Boolean()
|
||||
obj = doc.Dictionary()
|
||||
many = doc.Float()
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
'data,expected',
|
||||
(
|
||||
(
|
||||
dict(name='foo', count='4'),
|
||||
dict(name='foo', count=4),
|
||||
),
|
||||
(
|
||||
dict(name='foo', count=4, fish=['red', 'blue'], on='false'),
|
||||
dict(name='foo', count=4, fish=('red', 'blue'), on=False),
|
||||
),
|
||||
(
|
||||
dict(name='foo', count=4, fish=['red', 'blue'], on='true'),
|
||||
dict(name='foo', count=4, fish=('red', 'blue'), on=True),
|
||||
),
|
||||
(
|
||||
dict(name='foo', count=4, fish=['red', 'blue'], on='t'),
|
||||
dict(name='foo', count=4, fish=('red', 'blue'), on=True),
|
||||
),
|
||||
(
|
||||
dict(name='foo', count=4, fish=['red', 'blue'], on=1),
|
||||
dict(name='foo', count=4, fish=('red', 'blue'), on=True),
|
||||
),
|
||||
(
|
||||
dict(name='foo', count=1.5, obj={'foo': 'bar'}),
|
||||
dict(name='foo', count=1, obj={'foo': 'bar'}),
|
||||
),
|
||||
(
|
||||
dict(name='foo', count=1.5, many='10.3'),
|
||||
dict(name='foo', count=1, many=10.3),
|
||||
),
|
||||
)
|
||||
)
|
||||
def test_validate_doc(data, expected):
|
||||
model = Model()
|
||||
result = validate_data(model, data)
|
||||
assert result == expected
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
'data,expected',
|
||||
(
|
||||
(dict(), NoBodyContents),
|
||||
(dict(name='foo'), MissingRequiredField),
|
||||
(dict(name='foo', count=10, extra='too many'), ExcessJSONFields),
|
||||
)
|
||||
)
|
||||
def test_validate_doc_errors(data, expected):
|
||||
model = Model()
|
||||
pytest.raises(expected, validate_data, model, data)
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
'data,expected',
|
||||
(
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue