wrolpi/modules/archive/api.py
lrnselfreliance c23b014add
Feature/archive search (#70)
* Working through archive search.

* Using custom hook.

* Setting archive.contents duration creation.

* Archive search and domain list use custom hooks.

* Watching search param using value, not history.

* Search params are preserved on load.  Search and domain can be cleared.

* Fixing archive external link.

* Archive search form is new generic SearchInput.

* Reworking Archives empty states.

* Simplifying Archives logic.

* Fixing bug which displayed the wrong empty state during loading.

* Archives can be synced and deleted.

* Changing wording of archive delete.

* Removing URL table.

* Deleting domain if it has no archives.

* Fixing FE use of archive.url.  Fixing bug with get_domains.

* Universal results are displayed on Welcome page.

* Creating ExternalLink to open a link in a new tab.g

* Increasing size of search headers.

* Adding divider to ease reading.

* Adding More button below search results.

* Disabling More buttons when there are no results.

* Ordering videos by rank so linked results are at the top.

Signed-off-by: Roland <roland@learningselfreliance.com>
2022-01-03 15:13:38 -07:00

65 lines
1.7 KiB
Python

import asyncio
from http import HTTPStatus
from requests import Request
from sanic import response
from wrolpi.common import logger, wrol_mode_check
from wrolpi.errors import ValidationError
from wrolpi.root_api import get_blueprint, json_response
from wrolpi.schema import validate_doc, JSONErrorResponse
from . import lib
from .schema import ArchiveSearchRequest, ArchiveSearchResponse
NAME = 'archive'
bp = get_blueprint('Archive', '/api/archive')
logger = logger.getChild(__name__)
@bp.delete('/<archive_id:int>')
@validate_doc(
'Delete an individual archive',
responses=[
(HTTPStatus.NOT_FOUND, JSONErrorResponse),
]
)
@wrol_mode_check
async def delete_archive(_: Request, archive_id: int):
lib.delete_archive(archive_id)
return response.empty()
@bp.post('/refresh')
@wrol_mode_check
async def refresh_archives(_: Request):
asyncio.ensure_future(lib.refresh_archives())
return response.empty()
@bp.get('/domains')
async def fetch_domains(_: Request):
domains = lib.get_domains()
return json_response({'domains': domains, 'totals': {'domains': len(domains)}})
@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 = int(data.get('limit', 20))
offset = int(data.get('offset', 0))
except Exception as e:
raise ValidationError('Unable to validate search queries') from e
archives, total = lib.search(search_str, domain, limit, offset)
ret = dict(archives=archives, totals=dict(archives=total))
return json_response(ret)