from __future__ import annotations
from contextlib import contextmanager
from datetime import datetime
from functools import cached_property
from json import JSONDecodeError
from logging import getLogger
from logging import NullHandler
from onegov.api.form import model_from_form
from onegov.core.orm import Base
from onegov.user import User
from pydantic import ValidationError
from sqlalchemy import ForeignKey
from sqlalchemy.orm import mapped_column
from sqlalchemy.orm import relationship
from sqlalchemy.orm import Mapped
from uuid import uuid4, UUID
from webob.exc import HTTPException
from webob.multidict import MultiDict
from typing import TYPE_CHECKING, Any, ClassVar, NoReturn, Self, overload
if TYPE_CHECKING:
from collections.abc import Callable, Collection, Iterator, Mapping
from onegov.core import Framework
from onegov.core.collection import PKType
from onegov.core.request import CoreRequest
from onegov.form import Form
from sqlalchemy.orm import DeclarativeBase, Query, Session
from typing import Protocol
[docs]
log = getLogger('onegov.api')
log.addHandler(NullHandler())
[docs]
class ApiException(Exception):
"""Base class for all API exceptions.
Mainly used to ensure that all exceptions regarding the API are rendered
with the correct content type.
"""
def __init__(
self,
message: str = 'Internal Server Error',
status_code: int = 500,
headers: dict[str, str] | None = None,
):
super().__init__()
[docs]
self.status_code = status_code
@classmethod
@contextmanager
[docs]
def capture_exceptions(
cls,
default_message: str = 'Internal Server Error',
default_status_code: int = 500,
headers: dict[str, str] | None = None,
exception_type: type[Exception] = Exception,
) -> Iterator[None]:
try:
yield
except exception_type as exc:
# NOTE: log unexpected exceptions
if (
exception_type is Exception
and not isinstance(exc, (HTTPException, ApiException))
):
log.exception('Captured OneGov API Exception')
message = getattr(exc, 'message',
getattr(exc, 'title', default_message)
)
status_code = getattr(exc, 'status_code', default_status_code)
raise cls(message, status_code, headers) from exc
[docs]
class ApiInvalidParamException(ApiException):
def __init__(
self, message: str = 'Invalid Parameter', status_code: int = 400
):
[docs]
self.status_code = status_code
[docs]
class ApiEndpointItem[M: DeclarativeBase, IdT: PKType]:
""" A single instance of an item of a specific endpoint.
Passes all functionality to the specific API endpoint and is mainly used
for routing.
"""
def __init__(self, request: CoreRequest, endpoint: str, id: str):
[docs]
self.endpoint = endpoint
@cached_property
[docs]
def api_endpoint(self) -> ApiEndpoint[M, IdT] | None:
endpoint = ApiEndpointCollection(
self.request).endpoints.get(self.endpoint)
return endpoint if endpoint else None
@cached_property
[docs]
def item(self) -> M | None:
if self.api_endpoint:
return self.api_endpoint.by_id(self.id) # for. ex. ExtendedAgency
return None
@property
[docs]
def data(self) -> dict[str, Any] | None:
if self.api_endpoint and (item := self.item):
return self.api_endpoint.item_data(item)
return None
@property
[docs]
def links(self) -> dict[str, Any] | None:
if self.api_endpoint and (item := self.item):
return self.api_endpoint.item_links(item)
return None
[docs]
class ApiEndpoint[M: DeclarativeBase, IdT: PKType]:
""" An API endpoint.
API endpoints wrap collection and do some filter mapping.
To add a new endpoint, inherit from this class and provide the missing
functions and properties at the bottom. Note that the collection is
expected to be to provide the functionality of
``onegov.core.collection.Pagination``.
"""
[docs]
pk_type: Callable[[str], IdT]
def __init__(
self,
request: CoreRequest,
extra_parameters: dict[str, list[str]] | None = None,
page: int | None = None,
):
[docs]
self.page = int(page) if page else page
@cached_property
[docs]
def filters(self) -> Mapping[str, Collection[str] | str | None]:
""" A mapping of the available filter params to their corresponding
description or a collection of possible values. If possible values
are specified it is assumed that the filter can be specfied multiple
times.
The description is optional and should only be used for non-trivial
filters that don't just accept arbitrary strings.
"""
return {}
@property
[docs]
def title(self) -> str | None:
""" A human readable title for this endpoint. """
return None
@property
[docs]
def description(self) -> str | None:
""" A human readable description for this endpoint. """
return None
[docs]
def for_page(self, page: int | None) -> Self | None:
""" Return a new endpoint instance with the given page while keeping
the current filters.
"""
return self.__class__(self.request, self.extra_parameters, page)
[docs]
def for_filter(self, **filters: list[str]) -> Self:
""" Return a new endpoint instance with the given filters while
discarding the current filters and page.
"""
return self.__class__(self.request, filters)
@overload
[docs]
def for_item(self, item: None) -> None: ...
@overload
def for_item(self, item: M) -> ApiEndpointItem[M, IdT]: ...
def for_item(self, item: M | None) -> ApiEndpointItem[M, IdT] | None:
""" Return a new endpoint item instance with the given item. """
if not item:
return None
assert hasattr(item, 'id')
return self.for_item_id(item.id)
@overload
[docs]
def for_item_id(self, item_id: None) -> None: ...
@overload
def for_item_id(self, item_id: IdT) -> ApiEndpointItem[M, IdT]: ...
def for_item_id(
self,
item_id: IdT | None
) -> ApiEndpointItem[M, IdT] | None:
""" Return a new endpoint item instance with the given item id. """
if not item_id:
return None
if isinstance(item_id, int):
target = str(item_id)
elif isinstance(item_id, str):
target = item_id
else:
target = item_id.hex
return ApiEndpointItem(self.request, self.endpoint, target)
[docs]
def scalarize_value(
self,
name: str,
values: list[str] | None = None
) -> str | None:
if values is None:
values = self.extra_parameters.get(name)
if not values:
return None
if len(values) > 1:
raise ApiInvalidParamException(
f'Url parameter {name!r} may only be specified once.'
)
return values[0]
[docs]
def get_filter[T = str, DefaultT = None, EmptyT = None](
self,
name: str,
default: DefaultT = None, # type: ignore[assignment]
empty: EmptyT = None, # type: ignore[assignment]
coerce: Callable[[str], T] = str # type: ignore[assignment]
) -> T | DefaultT | EmptyT:
"""Returns the scalar filter value with the given name."""
if name not in self.extra_parameters:
return default
value = self.scalarize_value(name)
if value is None:
return empty
if coerce is not str:
try:
return coerce(value)
except Exception:
return default
return value # type: ignore[return-value]
[docs]
def by_id(self, id: IdT | str) -> M | None:
""" Return the item with the given ID from the collection. """
if self.pk_type is not str and isinstance(id, str):
try:
id = self.pk_type(id)
except Exception:
return None
return self.__class__(self.request).collection.by_id(id)
@property
[docs]
def session(self) -> Session:
return self.app.session()
@property
[docs]
def links(self) -> dict[str, Self | None]:
""" A dictionary with pagination instances. """
result: dict[str, Self | None] = {'prev': None, 'next': None}
previous = self.collection.previous
if previous:
result['prev'] = self.for_page(previous.page)
next_ = self.collection.next
if next_:
result['next'] = self.for_page(next_.page)
return result
@property
[docs]
def batch(self) -> dict[ApiEndpointItem[M, IdT], M]:
""" A dictionary with endpoint item instances and their titles. """
return {
self.for_item(item): item
for item in self.collection.batch
}
[docs]
def item_data(self, item: M) -> dict[str, Any]:
""" Return the data properties of the collection item as a dictionary.
For example::
{
'name': 'Paul',
'age': 40
}
"""
raise NotImplementedError()
[docs]
def item_links(self, item: M) -> dict[str, Any]:
""" Return the link properties of the collection item as a dictionary.
Links can either be string or a linkable object.
For example::
{
'website': 'https://onegov.ch',
'friends': FriendsApiEndpoint(app).for_item(paul),
'home': ApiEndpointCollection(app)
}
"""
raise NotImplementedError()
[docs]
def apply_changes(self, item: M, form: Any) -> dict[str, Any] | None:
""" Apply the changes to the item based on the given form data.
May optionally return a valid Collection+JSON payload which will
be included in the response.
"""
raise NotImplementedError()
@property
[docs]
def collection(self) -> PaginationWithById[M, Any]:
""" An instance of the collection with filters and page set. """
raise NotImplementedError()
[docs]
def assert_valid_filter(self, param: str) -> None:
if param not in self.filters:
raise ApiInvalidParamException(
f'Invalid url parameter {param!r}. Valid params are: '
f'{", ".join(sorted(self.filters))}')
# HACK: This gets around the fact that extra_parameters only
# supports scalar values, but we want to support lists
# of values for extra_parameters.
[docs]
def __link_alias__(self) -> str:
return self.request.class_link(
self.__class__,
{
'endpoint': self.endpoint,
'page': self.page,
},
query_params=MultiDict(
(key, value)
for key, values in self.extra_parameters.items()
for value in values
)
)
[docs]
class ApiEndpointCollection:
""" A collection of all available API endpoints. """
def __init__(self, request: CoreRequest):
@cached_property
[docs]
def endpoints(self) -> dict[str, ApiEndpoint[Any, Any]]:
settings = self.app.config.setting_registry
return {
endpoint.endpoint: endpoint
for endpoint in settings.api.endpoints(self.request)
}
[docs]
def get_endpoint(
self,
name: str,
page: int = 0,
extra_parameters: dict[str, list[str]] | None = None
) -> ApiEndpoint[Any, Any] | None:
endpoint = self.endpoints.get(name)
if endpoint is None:
return None
if extra_parameters:
endpoint = endpoint.for_filter(**extra_parameters)
if page:
endpoint = endpoint.for_page(page)
return endpoint
[docs]
class AuthEndpoint:
""" This is a Dummy, because morepath requires a model for linking. """
def __init__(self, app: Framework):
[docs]
class ApiKey(Base):
[docs]
__tablename__ = 'api_keys'
[docs]
id: Mapped[UUID] = mapped_column(
primary_key=True,
default=uuid4
)
#: the id of the user that created the api key
[docs]
user_id: Mapped[UUID] = mapped_column(ForeignKey('users.id'))
#: the user that created the api key
[docs]
user: Mapped[User] = relationship(back_populates='api_keys')
#: the name of the api key, may be any string
#: whether or not the api key can submit changes
[docs]
read_only: Mapped[bool] = mapped_column(default=True)
#: the last time a token was generated based on this api key
[docs]
last_used: Mapped[datetime | None]
#: the key itself
[docs]
key: Mapped[UUID] = mapped_column(default=uuid4)