3cd5034149
Some checks failed
Test execution / Test-all (push) Failing after 35s
This commit adds support for the new curated-write parameter `author_id`.
904 lines
35 KiB
Python
904 lines
35 KiB
Python
from __future__ import annotations
|
|
|
|
import logging
|
|
import re
|
|
from itertools import count
|
|
from typing import (
|
|
Callable,
|
|
Generator,
|
|
)
|
|
|
|
import requests
|
|
from requests import Session
|
|
from requests.exceptions import HTTPError
|
|
|
|
from . import JSON
|
|
|
|
|
|
__all__ = [
|
|
'HTTPError',
|
|
'JSON',
|
|
'get_session',
|
|
'get_paginated',
|
|
'get',
|
|
'collection_get_classes',
|
|
'collection_delete_record',
|
|
'collection_read_records',
|
|
'collection_read_records_of_class',
|
|
'collection_read_record_with_pid',
|
|
'collection_validate_record',
|
|
'collection_write_record',
|
|
'curated_delete_record',
|
|
'curated_read_records',
|
|
'curated_read_records_of_class',
|
|
'curated_read_record_with_pid',
|
|
'curated_write_record',
|
|
'incoming_delete_record',
|
|
'incoming_read_labels',
|
|
'incoming_read_records',
|
|
'incoming_read_records_of_class',
|
|
'incoming_read_record_with_pid',
|
|
'incoming_write_record',
|
|
'maintenance',
|
|
'server',
|
|
'Session',
|
|
]
|
|
|
|
|
|
logger = logging.getLogger('dump_things_pyclient')
|
|
|
|
|
|
def get_session() -> Session:
|
|
"""Return a session that can be used to reuse connections
|
|
|
|
:return: a Session-object that can be passed to most functions in this module
|
|
"""
|
|
return requests.Session()
|
|
|
|
|
|
def get_paginated(url: str,
|
|
token: str | None = None,
|
|
first_page: int = 1,
|
|
page_size: int = 100,
|
|
last_page: int | None = None,
|
|
parameters: dict[str, str] | None = None,
|
|
session: Session | None = None,
|
|
) -> Generator[tuple[JSON, int, int, int, int], None, None]:
|
|
"""Read all records from a paginated endpoint
|
|
|
|
:param url: URL of the paginated endpoint, e.g., `https://.../records/p/`
|
|
:param token: [optional] if str: token to authenticate against the endpoint,
|
|
if None: no token will be sent to the endpoint
|
|
:param first_page: [optional] first page to return (default: 1)
|
|
:param page_size: [optional] size of pages (default: 100)
|
|
:param last_page: [optional] last page to return (default: None (return all pages))
|
|
:param parameters: [optional] parameters to pass to the endpoint, the
|
|
parameter `page` is set automatically in this function
|
|
:param session: [optional] if set it will be used for making requests
|
|
|
|
:return: a Generator yielding tuples containing the current record, the
|
|
current page number, the total number of pages, the size of the pages,
|
|
and total number of records
|
|
"""
|
|
if last_page and last_page < first_page:
|
|
logger.warning('last_page (%d) < first_page (%d)', last_page, first_page)
|
|
return
|
|
|
|
for page in count(start=first_page):
|
|
result = _get_page(url, token, first_page=page, page_size=page_size, parameters=parameters, session=session)
|
|
total_pages, page_size, total_items = result['pages'], result['size'], result['total']
|
|
if total_pages == 0:
|
|
return
|
|
if last_page is None:
|
|
last_page = total_pages
|
|
|
|
yield from (
|
|
(record, page, total_pages, page_size, total_items)
|
|
for record in result['items'])
|
|
|
|
if page == min(last_page, total_pages):
|
|
return
|
|
|
|
|
|
def get(url: str,
|
|
token: str | None = None,
|
|
parameters: dict[str, str] | None = None,
|
|
session: Session | None = None,
|
|
) -> JSON:
|
|
"""Read JSON object from a non-paginated endpoint
|
|
|
|
:param url: URL of the endpoint, e.g., `https://.../records/`.
|
|
:param token: [optional] if str: token to authenticate against the endpoint,
|
|
if None: no token will be sent to the endpoint
|
|
:param parameters: [optional] parameters to pass to the endpoint
|
|
:param session: [optional] if set it will be used for making requests
|
|
|
|
:return: JSON object
|
|
"""
|
|
return _get_from_url(url, token, parameters, session)
|
|
|
|
|
|
def collection_get_classes(service_url: str,
|
|
collection: str,
|
|
session: Session | None = None,
|
|
) -> Generator[str, None, None]:
|
|
"""Read classes that are supported by the collection
|
|
|
|
Get the name of the classes that are known in the collection. If the
|
|
collection does not exist on the server, no class names are returned.
|
|
|
|
:param service_url: the base URL of the service, i.e., the URL up to
|
|
`/<collection>/...` or `/server`
|
|
:param collection: the name of the collection
|
|
:param session: [optional] if set it will be used for making requests
|
|
|
|
:return: a generator yielding names of the supported classes
|
|
"""
|
|
service_url = f'{service_url[:-1]}' if service_url.endswith('/') else service_url
|
|
matcher = re.compile(f'/{collection}/record/([A-Z][_a-zA-Z0-9]*)$')
|
|
open_api_spec = _get_from_url(
|
|
service_url + '/openapi.json',
|
|
token=None,
|
|
session=session,
|
|
)
|
|
for path in open_api_spec['paths']:
|
|
match = matcher.match(path)
|
|
if match:
|
|
yield match.group(1)
|
|
|
|
|
|
def collection_read_record_with_pid(service_url: str,
|
|
collection: str,
|
|
pid: str,
|
|
format: str = 'json',
|
|
token: str | None = None,
|
|
session: Session | None = None,
|
|
) -> dict | None:
|
|
"""Read record with the given pid from the collection on the service
|
|
|
|
Records are read from the curated area of the collection and from the
|
|
incoming area of the user identified by token, if a token is given.
|
|
Records from incoming areas take preference.
|
|
|
|
:param service_url: the base URL of the service, i.e., the URL up to
|
|
`/<collection>/...` or `/server`
|
|
:param collection: the name of the collection
|
|
:param pid: the PID of the record that should be retrieved
|
|
:param format: the format in which the result record should be returned,
|
|
either `json` or `ttl`
|
|
:param token: [optional] if set, a token to authenticate against
|
|
the endpoint, if None: no token will be sent to the endpoint
|
|
:param session: [optional] if set it will be used for making requests
|
|
|
|
:return: The record, if it exists, None otherwise.
|
|
"""
|
|
return get(
|
|
url=_build_url(service_url, collection, 'record'),
|
|
token=token,
|
|
parameters={'pid': pid, 'format': format},
|
|
session=session,
|
|
)
|
|
|
|
|
|
def collection_read_records(service_url: str,
|
|
collection: str,
|
|
matching: str | None = None,
|
|
format: str = 'json',
|
|
token: str | None = None,
|
|
page: int = 1,
|
|
size: int = 100,
|
|
last_page: int | None = None,
|
|
session: Session | None = None,
|
|
) -> Generator[tuple[dict, int, int, int, int], None, None]:
|
|
"""Read records from the collection on the service
|
|
|
|
:param service_url: the base URL of the service, i.e., the URL up to
|
|
`/<collection>/...` or `/server`
|
|
:param collection: the name of the collection
|
|
:param matching: [optional] return only records that have a matching value
|
|
(string comparison with `%` as wildcard)
|
|
:param format: the format in which the result records should be returned,
|
|
either `json` or `ttl` (default: `json`)
|
|
:param token: [optional] if set, a token to authenticate against
|
|
the endpoint, if None: no token will be sent to the endpoint.
|
|
:param page: int: the first page that should be returned (default: 1)
|
|
:param size: int: the number of records in an individual pages (default: 100)
|
|
:param last_page: int | None: if int, the last page that should be returned
|
|
if None, all pages following `page` will be returned
|
|
:param session: [optional] if set it will be used for making requests
|
|
|
|
:return: A generator yielding tuples containing: the current record, the
|
|
current page number, the total number of pages, the size of the
|
|
pages, the total number of records
|
|
"""
|
|
return get_paginated(
|
|
url=_build_url(service_url, collection, 'records/p/'),
|
|
token=token,
|
|
first_page=page,
|
|
page_size=size,
|
|
last_page=last_page,
|
|
parameters= {
|
|
'format': format,
|
|
**({'matching': matching} if matching else {})},
|
|
session=session,
|
|
)
|
|
|
|
|
|
def collection_read_records_of_class(
|
|
service_url: str,
|
|
collection: str,
|
|
class_name: str,
|
|
matching: str | None = None,
|
|
format: str = 'json',
|
|
token: str | None = None,
|
|
page: int = 1,
|
|
size: int = 100,
|
|
last_page: int | None = None,
|
|
session: Session | None = None,
|
|
) -> Generator[tuple[dict, int, int, int, int], None, None]:
|
|
"""Read records of the specified class from the collection on the service
|
|
|
|
:param service_url: the base URL of the service, i.e., the URL up to
|
|
`/<collection>/...` or `/server`
|
|
:param collection: the name of the collection
|
|
:param class_name: the name of the class whose instances should be returned
|
|
:param matching: [optional] return only records that have a matching value
|
|
(string comparison with `%` as wildcard)
|
|
:param format: the format in which the result records should be returned,
|
|
either `json` or `ttl` (default: `json`)
|
|
:param token: [optional] if set, a token to authenticate against
|
|
the endpoint, if None: no token will be sent to the endpoint.
|
|
:param page: int: the first page that should be returned (default: 1)
|
|
:param size: int: the number of records in an individual pages (default: 100)
|
|
:param last_page: int | None: if int, the last page that should be returned
|
|
if None, all pages following `page` will be returned
|
|
:param session: [optional] if set it will be used for making requests
|
|
|
|
:return: A generator yielding tuples containing: the current record, the
|
|
current page number, the total number of pages, the size of the
|
|
pages, the total number of records
|
|
"""
|
|
return get_paginated(
|
|
url=_build_url(service_url, collection, f'records/p/{class_name}'),
|
|
token=token,
|
|
first_page=page,
|
|
page_size=size,
|
|
last_page=last_page,
|
|
parameters= {
|
|
'format': format,
|
|
**({'matching': matching} if matching else {})},
|
|
session=session,
|
|
)
|
|
|
|
|
|
def collection_write_record(
|
|
service_url: str,
|
|
collection: str,
|
|
class_name: str,
|
|
record: dict | str,
|
|
format: str = 'json',
|
|
token: str | None = None,
|
|
session: Session | None = None,
|
|
) -> list[JSON]:
|
|
"""Write a record of the specified class to an inbox in the collection on the service
|
|
|
|
:param service_url: the base URL of the service, i.e., the URL up to
|
|
`/<collection>/...` or `/server`
|
|
:param collection: the name of the collection
|
|
:param class_name: the class of the record given in `record`
|
|
:param record: dict | str: the record that should be written
|
|
:param format: the format of `record`, either `json` or `ttl`
|
|
(default: `json`)
|
|
:param token: [optional] if set, a token to authenticate against
|
|
the endpoint, if None: no token will be sent to the endpoint
|
|
The token must have write access to incoming area in the collection
|
|
:param session: [optional] if set it will be used for making requests
|
|
|
|
:return list[JSON]: a list of records that was written. There might be more
|
|
than one record due to inlined-relations extraction. The individual
|
|
records might have annotations added
|
|
"""
|
|
_check_format_value(format)
|
|
return _post_to_url(
|
|
url=_build_url(service_url, collection, f'record/{class_name}'),
|
|
token=token,
|
|
params={'format': format},
|
|
session=session,
|
|
**(dict(json=record) if format == 'json' else dict(data=record)),
|
|
)
|
|
|
|
|
|
def collection_validate_record(
|
|
service_url: str,
|
|
collection: str,
|
|
class_name: str,
|
|
record: dict | str,
|
|
format: str = 'json',
|
|
token: str | None = None,
|
|
session: Session | None = None,
|
|
) -> list[JSON]:
|
|
"""Validate a record of the specified class in the collection on the service
|
|
|
|
Validation involves conversion of the record from json to ttl, or from
|
|
ttl to json.
|
|
|
|
:param service_url: the base URL of the service, i.e., the URL up to
|
|
`/<collection>/...` or `/server`
|
|
:param collection: the name of the collection
|
|
:param class_name: the class of the record given in `record`
|
|
:param record: dict | str: the record that should be validated
|
|
:param format: the format of `record`, either `json` or `ttl`
|
|
(default: `json`)
|
|
:param token: [optional] if set, a token to authenticate against
|
|
the endpoint, if None: no token will be sent to the endpoint
|
|
The token must have write access to incoming area in the collection
|
|
:param session: [optional] if set it will be used for making requests
|
|
|
|
:return: True
|
|
"""
|
|
_check_format_value(format)
|
|
return _post_to_url(
|
|
url=_build_url(service_url, collection, f'validate/{class_name}'),
|
|
token=token,
|
|
params={'format': format},
|
|
session=session,
|
|
**(dict(json=record) if format == 'json' else dict(data=record)),
|
|
)
|
|
|
|
|
|
def collection_delete_record(
|
|
service_url: str,
|
|
collection: str,
|
|
pid: str,
|
|
token: str | None = None,
|
|
session: Session | None = None,
|
|
) -> bool:
|
|
"""Delete the record with the given pid from the collection on the service
|
|
|
|
:param service_url: the base URL of the service, i.e., the URL up to
|
|
`/<collection>/...` or `/server`
|
|
:param collection: the name of the collection
|
|
:param pid: the PID of the record that should be deleted
|
|
:param token: [optional] if set, a token to authenticate against
|
|
the endpoint, if None: no token will be sent to the endpoint
|
|
:param session: [optional] if set it will be used for making requests
|
|
|
|
:return: True if the record was deleted, False otherwise
|
|
"""
|
|
return _delete_url(
|
|
url=_build_url(service_url, collection, 'record'),
|
|
token=token,
|
|
params={'pid': pid},
|
|
session=session,
|
|
)
|
|
|
|
|
|
def curated_read_record_with_pid(service_url: str,
|
|
collection: str,
|
|
pid: str,
|
|
token: str | None = None,
|
|
session: Session | None = None,
|
|
) -> dict | None:
|
|
"""Read record with the given pid from curated area of the collection on the service
|
|
|
|
The record will be returned as it is stored in the backend. That means
|
|
there is no "Schema-Type-Layer" involved.
|
|
|
|
:param service_url: the base URL of the service, i.e., the URL up to
|
|
`/<collection>/...` or `/server`
|
|
:param collection: the name of the collection
|
|
:param pid: the PID of the record that should be retrieved
|
|
:param token: [optional] if set, a token to authenticate against
|
|
the endpoint, if None: no token will be sent to the endpoint. A
|
|
token must have curator-rights
|
|
:param session: [optional] if set it will be used for making requests
|
|
|
|
:return: The record, if it exists, None otherwise
|
|
"""
|
|
return get(
|
|
url=_build_url(service_url, collection, 'curated/record'),
|
|
token=token,
|
|
parameters={'pid': pid},
|
|
session=session,
|
|
)
|
|
|
|
|
|
def curated_read_records(service_url: str,
|
|
collection: str,
|
|
matching: str | None = None,
|
|
token: str | None = None,
|
|
page: int = 1,
|
|
size: int = 100,
|
|
last_page: int | None = None,
|
|
session: Session | None = None,
|
|
) -> Generator[tuple[dict, int, int, int, int], None, None]:
|
|
"""Read records from the curated area the collection on the service
|
|
|
|
Records will be returned as they are stored in the backend. That means
|
|
there is no "Schema-Type-Layer" involved.
|
|
|
|
:param service_url: the base URL of the service, i.e., the URL up to
|
|
`/<collection>/...` or `/server`
|
|
:param collection: the name of the collection
|
|
:param matching: [optional] return only records that have a matching value
|
|
(string comparison with `%` as wildcard)
|
|
:param token: [optional] if set, a token to authenticate against
|
|
the endpoint, if None: no token will be sent to the endpoint. A
|
|
token must have curator-rights
|
|
:param page: int: the first page that should be returned (default: 1)
|
|
:param size: int: the number of records in an individual pages (default: 100)
|
|
:param last_page: int | None: if int, the last page that should be returned
|
|
if None, all pages following `page` will be returned
|
|
:param session: [optional] if set it will be used for making requests
|
|
|
|
:return: A generator yielding tuples containing: the current record, the
|
|
current page number, the total number of pages, the size of the
|
|
pages, the total number of records
|
|
"""
|
|
return get_paginated(
|
|
url=_build_url(service_url, collection, 'curated/records/p/'),
|
|
token=token,
|
|
first_page=page,
|
|
page_size=size,
|
|
last_page=last_page,
|
|
parameters={'matching': matching} if matching else {},
|
|
session=session,
|
|
)
|
|
|
|
|
|
def curated_read_records_of_class(
|
|
service_url: str,
|
|
collection: str,
|
|
class_name: str,
|
|
matching: str | None = None,
|
|
token: str | None = None,
|
|
page: int = 1,
|
|
size: int = 100,
|
|
last_page: int | None = None,
|
|
session: Session | None = None,
|
|
) -> Generator[tuple[dict, int, int, int, int], None, None]:
|
|
"""Read records of class `class_name` from the curated area the collection on the service
|
|
|
|
Records will be returned as they are stored in the backend. That means
|
|
there is no "Schema-Type-Layer" involved.
|
|
|
|
:param service_url: the base URL of the service, i.e., the URL up to
|
|
`/<collection>/...` or `/server`
|
|
:param collection: the name of the collection
|
|
:param class_name: the name of the class whose instances should be returned
|
|
:param matching: [optional] return only records that have a matching value
|
|
(string comparison with `%` as wildcard)
|
|
:param token: [optional] if set, a token to authenticate against
|
|
the endpoint, if None: no token will be sent to the endpoint. A
|
|
token must have curator-rights for the collection
|
|
:param page: int: the first page that should be returned (default: 1)
|
|
:param size: int: the number of records in an individual pages (default: 100)
|
|
:param last_page: int | None: if int, the last page that should be returned
|
|
if None, all pages following `page` will be returned
|
|
:param session: [optional] if set it will be used for making requests
|
|
|
|
:return: A generator yielding tuples containing: the current record, the
|
|
current page number, the total number of pages, the size of the
|
|
pages, the total number of records
|
|
"""
|
|
return get_paginated(
|
|
url=_build_url(service_url, collection, f'curated/records/p/{class_name}'),
|
|
token=token,
|
|
first_page=page,
|
|
page_size=size,
|
|
last_page=last_page,
|
|
parameters={'matching': matching} if matching else {},
|
|
session=session,
|
|
)
|
|
|
|
|
|
def curated_write_record(
|
|
service_url: str,
|
|
collection: str,
|
|
class_name: str,
|
|
record: dict,
|
|
author_id: str | None = None,
|
|
token: str | None = None,
|
|
session: Session | None = None,
|
|
) -> list[JSON]:
|
|
"""Write a record of the specified class to the curated area of the collection on the service
|
|
|
|
Records will be written without modification, i.e. there is no
|
|
"Schema-Type-Layer", there is no extraction of inlined records, and there
|
|
is no annotation-adding.
|
|
|
|
:param service_url: the base URL of the service, i.e., the URL up to
|
|
`/<collection>/...` or `/server`
|
|
:param collection: the name of the collection
|
|
:param class_name: the class of the record given in `record`
|
|
:param record: dict: the record that should be written
|
|
:param author_id: [optional] if set, send the given author ID to the
|
|
server, which will use it in audit logs. If not set, the server will
|
|
use the curator's ID as author ID.
|
|
:param token: [optional] if set, a token to authenticate against
|
|
the endpoint, if None: no token will be sent to the endpoint
|
|
A given token must have curator-rights for the collection
|
|
:param session: [optional] if set it will be used for making requests
|
|
|
|
:return list[JSON]: a list containing the record that was written
|
|
"""
|
|
return _post_to_url(
|
|
url=_build_url(service_url, collection, f'curated/record/{class_name}'),
|
|
token=token,
|
|
params={'author_id': author_id} if author_id else {},
|
|
session=session,
|
|
json=record,
|
|
)
|
|
|
|
|
|
def curated_delete_record(
|
|
service_url: str,
|
|
collection: str,
|
|
pid: str,
|
|
token: str | None = None,
|
|
session: Session | None = None,
|
|
) -> bool:
|
|
"""Delete the record with the given pid from the curated area of the collection on the service
|
|
|
|
:param service_url: the base URL of the service, i.e., the URL up to
|
|
`/<collection>/...` or `/server`
|
|
:param collection: the name of the collection
|
|
:param pid: the PID of the record that should be deleted
|
|
:param token: [optional] if set, a token to authenticate against
|
|
the endpoint, if None: no token will be sent to the endpoint
|
|
A given token must have curator-rights for the collection
|
|
:param session: [optional] if set it will be used for making requests
|
|
|
|
:return: True if the record was deleted, False otherwise
|
|
"""
|
|
return _delete_url(
|
|
url=_build_url(service_url, collection, 'curated/record'),
|
|
token=token,
|
|
params={'pid': pid},
|
|
session=session,
|
|
)
|
|
|
|
|
|
def incoming_read_labels(service_url: str,
|
|
collection: str,
|
|
token: str | None = None,
|
|
session: Session | None = None,
|
|
) -> Generator[str, None, None]:
|
|
"""Read all incoming labels for the collection on the service.
|
|
|
|
:param service_url: the base URL of the service, i.e., the URL up to
|
|
`/<collection>/...` or `/server`
|
|
:param collection: the name of the collection
|
|
:param token: [optional] if set, a token to authenticate against
|
|
the endpoint, if None: no token will be sent to the endpoint
|
|
A given token must have curator-rights for the collection
|
|
:param session: [optional] if set it will be used for making requests
|
|
|
|
:return: list[str]: a list of incoming area labels
|
|
"""
|
|
yield from _get_from_url(
|
|
url=_build_url(service_url, collection,'incoming/'),
|
|
token=token,
|
|
session=session,
|
|
)
|
|
|
|
|
|
def incoming_read_record_with_pid(service_url: str,
|
|
collection: str,
|
|
label: str,
|
|
pid: str,
|
|
token: str | None = None,
|
|
session: Session | None = None,
|
|
) -> dict | None:
|
|
"""Read record with the given pid from the specified incoming area of the collection on the service
|
|
|
|
The record will be returned as it is stored in the backend. That means
|
|
there is no "Schema-Type-Layer" involved.
|
|
|
|
:param service_url: the base URL of the service, i.e., the URL up to
|
|
`/<collection>/...` or `/server`
|
|
:param collection: the name of the collection
|
|
:param label: the label of the incoming area in the collection
|
|
:param pid: the PID of the record that should be retrieved
|
|
:param token: [optional] if set, a token to authenticate against
|
|
the endpoint, if None: no token will be sent to the endpoint. A
|
|
token must have curator-rights
|
|
:param session: [optional] if set it will be used for making requests
|
|
|
|
:return: The record, if it exists, None otherwise
|
|
"""
|
|
return get(
|
|
url=_build_incoming_url(service_url, collection, label, 'record'),
|
|
token=token,
|
|
parameters={'pid': pid},
|
|
session=session,
|
|
)
|
|
|
|
|
|
def incoming_read_records(service_url: str,
|
|
collection: str,
|
|
label: str,
|
|
matching: str | None = None,
|
|
token: str | None = None,
|
|
page: int = 1,
|
|
size: int = 100,
|
|
last_page: int | None = None,
|
|
session: Session | None = None,
|
|
) -> Generator[tuple[dict, int, int, int, int], None, None]:
|
|
"""Read records from the specified incoming area the collection on the service
|
|
|
|
Records will be returned as they are stored in the backend. That means
|
|
there is no "Schema-Type-Layer" involved.
|
|
|
|
:param service_url: the base URL of the service, i.e., the URL up to
|
|
`/<collection>/...` or `/server`
|
|
:param collection: the name of the collection
|
|
:param label: the label of the incoming area in the collection
|
|
:param matching: [optional] return only records that have a matching value
|
|
(string comparison with `%` as wildcard)
|
|
:param token: [optional] if set, a token to authenticate against
|
|
the endpoint, if None: no token will be sent to the endpoint. A
|
|
token must have curator-rights for the collection
|
|
:param page: int: the first page that should be returned (default: 1)
|
|
:param size: int: the number of records in an individual pages (default: 100)
|
|
:param last_page: int | None: if int, the last page that should be returned
|
|
if None, all pages following `page` will be returned
|
|
:param session: [optional] if set it will be used for making requests
|
|
|
|
:return: A generator yielding tuples containing: the current record, the
|
|
current page number, the total number of pages, the size of the
|
|
pages, the total number of records
|
|
"""
|
|
return get_paginated(
|
|
url=_build_incoming_url(service_url, collection, label,'records/p/'),
|
|
token=token,
|
|
first_page=page,
|
|
page_size=size,
|
|
last_page=last_page,
|
|
parameters={'matching': matching} if matching else {},
|
|
session=session,
|
|
)
|
|
|
|
|
|
def incoming_read_records_of_class(
|
|
service_url: str,
|
|
collection: str,
|
|
label: str,
|
|
class_name: str,
|
|
matching: str | None = None,
|
|
token: str | None = None,
|
|
page: int = 1,
|
|
size: int = 100,
|
|
last_page: int | None = None,
|
|
session: Session | None = None,
|
|
) -> Generator[tuple[dict, int, int, int, int], None, None]:
|
|
"""Read records of the specified class from the specified incoming area the collection on the service
|
|
|
|
Records will be returned as they are stored in the backend. That means
|
|
there is no "Schema-Type-Layer" involved.
|
|
|
|
:param service_url: the base URL of the service, i.e., the URL up to
|
|
`/<collection>/...` or `/server`
|
|
:param collection: the name of the collection
|
|
:param label: the label of the incoming area in the collection
|
|
:param class_name: the name of the class whose instances should be returned
|
|
:param matching: [optional] return only records that have a matching value
|
|
(string comparison with `%` as wildcard)
|
|
:param token: [optional] if set, a token to authenticate against
|
|
the endpoint, if None: no token will be sent to the endpoint. A
|
|
token must have curator-rights for the collection
|
|
:param page: int: the first page that should be returned (default: 1)
|
|
:param size: int: the number of records in an individual pages (default: 100)
|
|
:param last_page: int | None: if int, the last page that should be returned
|
|
if None, all pages following `page` will be returned
|
|
:param session: [optional] if set it will be used for making requests
|
|
|
|
:return: A generator yielding tuples containing: the current record, the
|
|
current page number, the total number of pages, the size of the
|
|
pages, the total number of records
|
|
"""
|
|
return get_paginated(
|
|
url=_build_incoming_url(service_url, collection, label,f'records/p/{class_name}'),
|
|
token=token,
|
|
first_page=page,
|
|
page_size=size,
|
|
last_page=last_page,
|
|
parameters={'matching': matching} if matching else {},
|
|
session=session,
|
|
)
|
|
|
|
|
|
def incoming_write_record(
|
|
service_url: str,
|
|
collection: str,
|
|
label: str,
|
|
class_name: str,
|
|
record: dict,
|
|
token: str | None = None,
|
|
session: Session | None = None,
|
|
) -> list[JSON]:
|
|
"""Write a record of the specified class to the specified incoming area of the collection on the service
|
|
|
|
Records will be written without modification, i.e. there is no
|
|
"Schema-Type-Layer", there is no extraction of inlined records, and there
|
|
is no annotation-adding.
|
|
|
|
:param service_url: the base URL of the service, i.e., the URL up to
|
|
`/<collection>/...` or `/server`
|
|
:param collection: the name of the collection
|
|
:param label: the label of the incoming area in the collection
|
|
:param class_name: the class of the record given in `record`
|
|
:param record: dict: the record that should be written
|
|
:param token: [optional] if set, a token to authenticate against
|
|
the endpoint, if None: no token will be sent to the endpoint
|
|
A given token must have curator-rights for the collection
|
|
:param session: [optional] if set it will be used for making requests
|
|
|
|
:return list[JSON]: a list containing the record that was written
|
|
"""
|
|
return _post_to_url(
|
|
url=_build_incoming_url(service_url, collection, label, f'record/{class_name}'),
|
|
token=token,
|
|
json=record,
|
|
session=session,
|
|
)
|
|
|
|
|
|
def incoming_delete_record(
|
|
service_url: str,
|
|
collection: str,
|
|
label: str,
|
|
pid: str,
|
|
token: str | None = None,
|
|
session: Session | None = None,
|
|
) -> bool:
|
|
"""Delete the record with the given pid from the specified incoming area of the collection on the service
|
|
|
|
:param service_url: the base URL of the service, i.e., the URL up to
|
|
`/<collection>/...` or `/server`
|
|
:param collection: the name of the collection
|
|
:param label: the label of the incoming area in the collection
|
|
:param pid: the PID of the record that should be deleted
|
|
:param token: [optional] if set, a token to authenticate against
|
|
the endpoint, if None: no token will be sent to the endpoint
|
|
A given token must have curator-rights for the collection
|
|
:param session: [optional] if set, it will be used for requests
|
|
|
|
:return: True if the record was deleted, False otherwise
|
|
"""
|
|
return _delete_url(
|
|
url=_build_incoming_url(service_url, collection, label,'record'),
|
|
token=token,
|
|
params={'pid': pid},
|
|
session=session,
|
|
)
|
|
|
|
|
|
def server(
|
|
service_url: str,
|
|
session: Session | None = None,
|
|
) -> JSON:
|
|
"""Get server-information from the service
|
|
|
|
:param service_url: the base URL of the service, i.e., the URL up to
|
|
`/<collection>/...` or `/server`
|
|
:param session: an optional requests.Session object to use for making requests
|
|
|
|
:return: information returned by the `<service_url>/server` endpoint
|
|
"""
|
|
url = (
|
|
(f'{service_url[:-1]}' if service_url.endswith('/') else service_url)
|
|
+ '/server'
|
|
)
|
|
method = session.get if session else requests.get
|
|
return _do_request(method, url=url, token=None, params=None)
|
|
|
|
|
|
def maintenance(
|
|
service_url: str,
|
|
collection: str,
|
|
active: bool,
|
|
token: str,
|
|
session: Session | None = None,
|
|
) -> None:
|
|
"""Activate or deactivate maintenance mode of a collection
|
|
|
|
:param service_url: the base URL of the service, i.e., the URL up to
|
|
`/<collection>/...`, `/maintenance`, or `/server`
|
|
:param collection: the name of the collection
|
|
:param active: whether maintenance mode should be active (`True`) or
|
|
non-active (`False`).
|
|
:param token: a token to authenticate against the endpoint, the token
|
|
must have curator-rights for the collection
|
|
:param session: an optional requests.Session object to use for making requests
|
|
"""
|
|
url = (
|
|
(f'{service_url[:-1]}' if service_url.endswith('/') else service_url)
|
|
+ '/maintenance'
|
|
)
|
|
_post_to_url(
|
|
url=url,
|
|
token=token,
|
|
session=session,
|
|
json={'collection': collection, 'active': active},
|
|
)
|
|
|
|
|
|
def _get_from_url(url: str,
|
|
token: str | None,
|
|
params: dict[str, str] | None = None,
|
|
session: Session | None = None,
|
|
) -> JSON:
|
|
method = session.get if session else requests.get
|
|
return _do_request(method, url, token, params=params)
|
|
|
|
|
|
def _post_to_url(url: str,
|
|
token: str | None,
|
|
params: dict[str, str] | None = None,
|
|
session: Session | None = None,
|
|
**kwargs,
|
|
) -> JSON:
|
|
method = session.post if session else requests.post
|
|
return _do_request(method, url, token, params=params, **kwargs)
|
|
|
|
|
|
def _delete_url(url: str,
|
|
token: str | None,
|
|
params: dict[str, str] | None = None,
|
|
session: Session | None = None,
|
|
) -> JSON:
|
|
method = session.delete if session else requests.delete
|
|
return _do_request(method, url, token, params=params)
|
|
|
|
|
|
def _do_request(method: Callable,
|
|
url: str,
|
|
token: str | object | None,
|
|
params: dict[str, str] | None,
|
|
**kwargs,
|
|
) -> JSON:
|
|
headers = {'x-dumpthings-token': token} if token is not None else {}
|
|
response = method(url, headers=headers, params=params or {}, **kwargs)
|
|
response.raise_for_status()
|
|
if response.headers.get('content-type', '').strip().startswith('text/turtle'):
|
|
return response.text
|
|
return response.json()
|
|
|
|
|
|
def _build_url(
|
|
service_url: str,
|
|
collection: str,
|
|
tail: str,
|
|
) -> str:
|
|
service_url = f'{service_url[:-1]}' if service_url.endswith('/') else service_url
|
|
collection = f'{collection[:-1]}' if collection.endswith('/') else collection
|
|
return f'{service_url}/{collection}/{tail}'
|
|
|
|
|
|
def _build_incoming_url(
|
|
service_url: str,
|
|
collection: str,
|
|
label: str,
|
|
tail: str,
|
|
) -> str:
|
|
label = f'{label[:-1]}' if label.endswith('/') else label
|
|
return _build_url(service_url, collection, f'incoming/{label}/{tail}')
|
|
|
|
|
|
def _get_page(url_base: str,
|
|
token: str | None = None,
|
|
first_page: int = 1,
|
|
page_size: int = 100,
|
|
parameters: dict | None = None,
|
|
session: Session | None = None,
|
|
) -> JSON:
|
|
parameters = parameters or {}
|
|
parameters['page'] = first_page
|
|
parameters['size'] = page_size
|
|
return _get_from_url(url_base, token, parameters, session)
|
|
|
|
|
|
def _check_format_value(format: str) -> None:
|
|
if format not in ('json', 'ttl'):
|
|
raise ValueError('Format must be either "json" or "ttl"')
|