Source code for piccolo_admin.endpoints

"""
Creates a basic wrapper around a Piccolo model, turning it into an ASGI app.
"""

from __future__ import annotations

import inspect
import itertools
import json
import logging
import os
import typing as t
from dataclasses import dataclass
from datetime import timedelta
from functools import partial

from fastapi import FastAPI, File, Form, UploadFile
from piccolo.apps.user.tables import BaseUser
from piccolo.columns.base import Column
from piccolo.columns.column_types import (
    ForeignKey,
    Time,
    Timestamp,
    Timestamptz,
)
from piccolo.columns.reference import LazyTableReference
from piccolo.table import Table
from piccolo.utils.warnings import Level, colored_warning
from piccolo_api.change_password.endpoints import change_password
from piccolo_api.crud.endpoints import OrderBy, PiccoloCRUD
from piccolo_api.crud.hooks import Hook
from piccolo_api.crud.validators import Validators
from piccolo_api.csrf.middleware import CSRFMiddleware
from piccolo_api.fastapi.endpoints import FastAPIKwargs, FastAPIWrapper
from piccolo_api.media.base import MediaStorage
from piccolo_api.media.local import LocalMediaStorage
from piccolo_api.openapi.endpoints import swagger_ui
from piccolo_api.rate_limiting.middleware import (
    InMemoryLimitProvider,
    RateLimitingMiddleware,
    RateLimitProvider,
)
from piccolo_api.session_auth.endpoints import session_login, session_logout
from piccolo_api.session_auth.middleware import SessionsAuthBackend
from piccolo_api.session_auth.tables import SessionsBase
from pydantic import BaseModel, Field, ValidationError
from starlette.middleware import Middleware
from starlette.middleware.authentication import AuthenticationMiddleware
from starlette.middleware.exceptions import HTTPException
from starlette.requests import Request
from starlette.responses import HTMLResponse, JSONResponse
from starlette.staticfiles import StaticFiles

from .translations.data import TRANSLATIONS
from .translations.models import (
    Translation,
    TranslationListItem,
    TranslationListResponse,
)
from .version import __VERSION__ as PICCOLO_ADMIN_VERSION

logger = logging.getLogger(__name__)


ASSET_PATH = os.path.join(os.path.dirname(__file__), "dist")


class UserResponseModel(BaseModel):
    username: str
    user_id: str


class MetaResponseModel(BaseModel):
    piccolo_admin_version: str
    site_name: str


class StoreFileResponseModel(BaseModel):
    file_key: str = Field(description="For example `my_file-some-uuid.jpeg`.")


class GenerateFileURLRequestModel(BaseModel):
    column_name: str
    table_name: str
    file_key: str = Field(description="For example `my_file-some-uuid.jpeg`.")


class GenerateFileURLResponseModel(BaseModel):
    file_url: str = Field(description="A URL which the file is accessible on.")


class GroupedTableNamesResponseModel(BaseModel):
    grouped: t.Dict[str, t.List[str]] = Field(default_factory=list)
    ungrouped: t.List[str] = Field(default_factory=list)


[docs]@dataclass class TableConfig: """ Gives the user more control over how a ``Table`` appears in the UI. :param table_class: The :class:`Table <piccolo.table.Table>` class to configure. :param visible_columns: If specified, only these columns will be shown in the list view of the UI. This is useful when you have a lot of columns. :param exclude_visible_columns: You can specify this instead of ``visible_columns``, in which case all of the ``Table`` columns except the ones specified will be shown in the list view. :param visible_filters: If specified, only these columns will be shown in the filter sidebar of the UI. This is useful when you have a lot of columns. :param exclude_visible_filters: You can specify this instead of ``visible_filters``, in which case all of the ``Table`` columns except the ones specified will be shown in the filter sidebar. :param rich_text_columns: You can specify ``rich_text_columns`` if you want a WYSIWYG editor on certain Piccolo :class:`Text <piccolo.columns.column_types.Text>` columns. Any columns not specified will use a standard HTML textarea tag in the UI. :param hooks: These are passed directly to :class:`PiccoloCRUD <piccolo_api.crud.endpoints.PiccoloCRUD>`, which powers Piccolo Admin under the hood. It allows you to run custom logic when a row is modified. :param media_storage: These columns will be used to store media. We don't directly store the media in the database, but instead store a string, which is a unique identifier, and can be used to retrieve a URL for accessing the file. Piccolo Admin automatically renders a file upload widget for each media column in the UI. :param validators: The :class:`Validators <piccolo_api.crud.endpoints.Validators>` are passed directly to :class:`PiccoloCRUD <piccolo_api.crud.endpoints.PiccoloCRUD>`, which powers Piccolo Admin under the hood. It allows fine grained access control over each API endpoint. For example, limiting which users can ``POST`` data:: from piccolo_api.crud.endpoints import PiccoloCRUD from starlette.exceptions import HTTPException from starlette.requests import Request async def manager_only( piccolo_crud: PiccoloCRUD, request: Request ): # The Piccolo `BaseUser` can be accessed from the request. user = request.user.user # Assuming we have another database table where we record # users with certain permissions. manager = await Manager.exists().where(manager.user == user) if not manager: # Raise a Starlette exception if we want to reject the # request. raise HTTPException( status_code=403, detail="Only managers are allowed to do this" ) admin = create_admin( tables=TableConfig( Movie, validators=Validators(post_single=[manager_only]) ) ) :param menu_group: If specified, tables can be divided into groups in the table menu. This is useful when you have many tables that you can organize into groups for better visibility. :param link_column: In the list view of Piccolo Admin, we use the primary key to link to the edit page. However, if the primary key column is hidden, due to ``visible_columns`` or ``exclude_visible_columns``, then we need to specify an alternative column to use as the link. :param order_by: If specified, the rows are sorted by ``order_by``, otherwise the default ``primary_key`` column is used to sort the rows. :param time_resolution: Controls the resolution of ``Time`` columns, and the time component of ``Timestamp`` / ``Timestamptz`` columns. The units are given in seconds. Some examples: * 0.001 - the max resolution is 1 millisecond (this is the minimum currently allowed by HTML input fields) * 1 - the max resolution is 1 second (the default) * 60 - the max resolution is 1 minute """ table_class: t.Type[Table] visible_columns: t.Optional[t.List[Column]] = None exclude_visible_columns: t.Optional[t.List[Column]] = None visible_filters: t.Optional[t.List[Column]] = None exclude_visible_filters: t.Optional[t.List[Column]] = None rich_text_columns: t.Optional[t.List[Column]] = None hooks: t.Optional[t.List[Hook]] = None media_storage: t.Optional[t.Sequence[MediaStorage]] = None validators: t.Optional[Validators] = None menu_group: t.Optional[str] = None link_column: t.Optional[Column] = None order_by: t.Optional[t.List[OrderBy]] = None time_resolution: t.Optional[ t.Dict[t.Union[Timestamp, Timestamptz, Time], t.Union[float, int]] ] = None def __post_init__(self): if self.visible_columns and self.exclude_visible_columns: raise ValueError( "Only specify `visible_columns` or " "`exclude_visible_columns`." ) if self.visible_filters and self.exclude_visible_filters: raise ValueError( "Only specify `visible_filters` or `exclude_visible_filters`." ) if isinstance(self.link_column, ForeignKey): raise ValueError( "Don't use a foreign key column for `link_column`, as they " "are already displayed as a link in the UI." ) # Create a mapping for faster lookups self.media_columns = ( {i.column: i for i in self.media_storage} if self.media_storage else None ) def _get_columns( self, include_columns: t.Optional[t.List[Column]], exclude_columns: t.Optional[t.List[Column]], all_columns: t.List[Column], ) -> t.List[Column]: if include_columns and not exclude_columns: return include_columns if exclude_columns and not include_columns: column_names = [i._meta.name for i in exclude_columns] return [i for i in all_columns if i._meta.name not in column_names] return all_columns def get_visible_columns(self) -> t.List[Column]: return self._get_columns( include_columns=self.visible_columns, exclude_columns=self.exclude_visible_columns, all_columns=self.table_class._meta.columns, ) def get_visible_column_names(self) -> t.Tuple[str, ...]: return tuple(i._meta.name for i in self.get_visible_columns()) def get_visible_filters(self) -> t.List[Column]: return self._get_columns( include_columns=self.visible_filters, exclude_columns=self.exclude_visible_filters, all_columns=self.table_class._meta.columns, ) def get_visible_filter_names(self) -> t.Tuple[str, ...]: return tuple(i._meta.name for i in self.get_visible_filters()) def get_rich_text_columns_names(self) -> t.Tuple[str, ...]: return ( tuple(i._meta.name for i in self.rich_text_columns) if self.rich_text_columns else () ) def get_media_columns_names(self) -> t.Tuple[str, ...]: return ( tuple(i._meta.name for i in self.media_columns) if self.media_columns else () ) def get_link_column(self) -> Column: return self.link_column or self.table_class._meta.primary_key def get_order_by(self) -> t.List[OrderBy]: return self.order_by or [ OrderBy(column=self.table_class._meta.primary_key, ascending=True) ] def get_time_resolution(self) -> t.Dict[str, t.Union[int, float]]: return ( { column._meta.name: resolution for column, resolution in self.time_resolution.items() } if self.time_resolution else {} )
PydanticModel = t.TypeVar("PydanticModel", bound=BaseModel)
[docs]@dataclass class FormConfig: """ Used to specify forms, which are passed into ``create_admin``. :param name: This will be displayed in the UI in the sidebar. :param pydantic_model: This determines which fields to display in the form, and is used to deserialise the responses. :param endpoint: Your custom handler, which accepts two arguments - the FastAPI / Starlette request object, in case you want to access the cookies / headers / logged in user (via `request.user`). And secondly an instance of the Pydantic model. If it returns a string, it will be shown to the user in the UI as the success message. For example ``'Successfully sent email'``. The endpoint can be a normal function or async function. :param description: An optional description which is shown in the UI to explain to the user what the form is for. Here's a full example: .. code-block:: python class MyModel(pydantic.BaseModel): message: str = "hello world" def my_endpoint(request: Request, data: MyModel): print(f"I received {data.message}") # If we're not happy with the data raise a ValueError # The message inside the exception will be displayed in the UI. raise ValueError("We were unable to process the form.") # If we're happy with the data, just return a string, which # will be displayed in the UI. return "Successful." config = FormConfig( name="My Form", pydantic_model=MyModel, endpoint=my_endpoint ) """ def __init__( self, name: str, pydantic_model: t.Type[PydanticModel], endpoint: t.Callable[ [Request, PydanticModel], t.Union[str, None, t.Coroutine], ], description: t.Optional[str] = None, ): self.name = name self.pydantic_model = pydantic_model self.endpoint = endpoint self.description = description self.slug = self.name.replace(" ", "-").lower()
class FormConfigResponseModel(BaseModel): name: str slug: str description: t.Optional[str] = None def handle_auth_exception(request: Request, exc: Exception): return JSONResponse({"error": "Auth failed"}, status_code=401) def superuser_validators(piccolo_crud: PiccoloCRUD, request: Request): """ We need to provide extra validation on certain tables (e.g. user and sessions), so only superusers can perform certain actions, otherwise the security of the application can be compromised. """ user: BaseUser = request.user.user if user.superuser: return if request.method.upper() in ["PUT", "PATCH", "DELETE", "POST"]: raise HTTPException( status_code=405, detail="Only superusers can perform these actions.", ) async def log_error(request: Request, exc: HTTPException): logger.exception("Server error") raise exc class AdminRouter(FastAPI): """ The root returns a single page app. The other URLs are REST endpoints. """ table: t.List[Table] = [] auth_table: t.Type[BaseUser] = BaseUser template: str = "" def __init__( self, *tables: t.Union[t.Type[Table], TableConfig], forms: t.List[FormConfig] = [], auth_table: t.Type[BaseUser] = BaseUser, session_table: t.Type[SessionsBase] = SessionsBase, session_expiry: timedelta = timedelta(hours=1), max_session_expiry: timedelta = timedelta(days=7), increase_expiry: t.Optional[timedelta] = timedelta(minutes=20), page_size: int = 15, read_only: bool = False, rate_limit_provider: t.Optional[RateLimitProvider] = None, production: bool = False, site_name: str = "Piccolo Admin", default_language_code: str = "auto", translations: t.Optional[t.List[Translation]] = None, allowed_hosts: t.Sequence[str] = [], debug: bool = False, sidebar_links: t.Dict[str, str] = {}, ) -> None: super().__init__( title=site_name, description=f"{site_name} documentation", middleware=[ Middleware(CSRFMiddleware, allowed_hosts=allowed_hosts) ], debug=debug, exception_handlers={500: log_error}, docs_url=None, redoc_url=None, ) ####################################################################### # Convert any table arguments which are plain ``Table`` classes into # ``TableConfig`` instances. table_configs: t.List[TableConfig] = [] for table in tables: if isinstance(table, TableConfig): table_configs.append(table) else: table_configs.append(TableConfig(table_class=table)) self.table_configs = sorted( table_configs, key=lambda table_config: table_config.table_class._meta.tablename, ) self.table_config_map = { table_config.table_class._meta.tablename: table_config for table_config in self.table_configs } ####################################################################### # Make sure columns are configured properly. for table_config in table_configs: table_class = table_config.table_class for column in table_class._meta.columns: if column._meta.secret and column._meta.required: message = ( f"{table_class._meta.tablename}." f"{column._meta._name} is using `secret` and " f"`required` column args which are incompatible. " f"You may encounter unexpected behavior when using " f"this table within Piccolo Admin." ) colored_warning(message, level=Level.high) ####################################################################### # Make sure media storage is configured properly. media_storage = [ i for i in itertools.chain( *[ table_config.media_storage or [] for table_config in table_configs ] ) ] if len(media_storage) != len(set(media_storage)): raise ValueError( "Media storage is misconfigured - multiple columns are saving " "to the same location." ) ####################################################################### self.default_language_code = default_language_code self.translations_map = { translation.language_code.lower(): translation for translation in (translations or TRANSLATIONS) } ####################################################################### self.auth_table = auth_table self.site_name = site_name self.forms = forms self.read_only = read_only self.sidebar_links = sidebar_links self.form_config_map = {form.slug: form for form in self.forms} with open(os.path.join(ASSET_PATH, "index.html")) as f: self.template = f.read() ####################################################################### private_app = FastAPI( docs_url=None, redoc_url=None, debug=debug, exception_handlers={500: log_error}, ) private_app.mount("/docs/", swagger_ui(schema_url="../openapi.json")) for table_config in table_configs: table_class = table_config.table_class visible_column_names = table_config.get_visible_column_names() visible_filter_names = table_config.get_visible_filter_names() rich_text_columns_names = ( table_config.get_rich_text_columns_names() ) media_columns_names = table_config.get_media_columns_names() link_column_name = table_config.get_link_column()._meta.name order_by = table_config.get_order_by() time_resolution = table_config.get_time_resolution() validators = table_config.validators if table_class in (auth_table, session_table): validators = validators or Validators() validators.every = [superuser_validators, *validators.every] FastAPIWrapper( root_url=f"/tables/{table_class._meta.tablename}/", fastapi_app=private_app, piccolo_crud=PiccoloCRUD( table=table_class, read_only=read_only, page_size=page_size, schema_extra={ "visible_column_names": visible_column_names, "visible_filter_names": visible_filter_names, "rich_text_columns": rich_text_columns_names, "media_columns": media_columns_names, "link_column_name": link_column_name, "order_by": tuple(i.to_dict() for i in order_by), "time_resolution": time_resolution, }, validators=validators, hooks=table_config.hooks, ), fastapi_kwargs=FastAPIKwargs( all_routes={ "tags": [f"{table_class._meta.tablename.capitalize()}"] }, ), ) private_app.add_api_route( path="/tables/", endpoint=self.get_table_list, # type: ignore methods=["GET"], response_model=t.List[str], tags=["Tables"], ) private_app.add_api_route( path="/tables/grouped/", endpoint=self.get_table_list_grouped, # type: ignore methods=["GET"], response_model=GroupedTableNamesResponseModel, tags=["Tables"], ) private_app.add_api_route( path="/links/", endpoint=self.get_sidebar_links, # type: ignore methods=["GET"], tags=["Links"], ) private_app.add_api_route( path="/forms/", endpoint=self.get_forms, # type: ignore methods=["GET"], tags=["Forms"], response_model=t.List[FormConfigResponseModel], ) private_app.add_api_route( path="/forms/{form_slug:str}/", endpoint=self.get_single_form, # type: ignore methods=["GET"], tags=["Forms"], ) private_app.add_api_route( path="/forms/{form_slug:str}/schema/", endpoint=self.get_single_form_schema, # type: ignore methods=["GET"], tags=["Forms"], ) private_app.add_api_route( path="/forms/{form_slug:str}/", endpoint=self.post_single_form, # type: ignore methods=["POST"], tags=["Forms"], ) private_app.add_api_route( path="/user/", endpoint=self.get_user, # type: ignore methods=["GET"], tags=["User"], response_model=UserResponseModel, ) private_app.add_route( path="/change-password/", route=change_password( login_url="./../../public/login/", session_table=session_table, read_only=read_only, ), methods=["POST"], ) ####################################################################### # Media private_app.add_api_route( path="/media/", endpoint=self.store_file, # type: ignore methods=["POST"], tags=["Media"], response_model=StoreFileResponseModel, ) private_app.add_api_route( path="/media/generate-file-url/", endpoint=self.generate_file_url, # type: ignore methods=["POST"], tags=["Media"], response_model=GenerateFileURLResponseModel, ) for table_config in self.table_configs: if table_config.media_columns: for ( column, media_storage, ) in table_config.media_columns.items(): if isinstance(media_storage, LocalMediaStorage): private_app.mount( path=f"/media-files/{column._meta.table._meta.tablename}/{column._meta.name}/", # noqa: E501 app=StaticFiles( directory=media_storage.media_path ), ) ####################################################################### public_app = FastAPI( redoc_url=None, docs_url=None, debug=debug, exception_handlers={500: log_error}, ) public_app.mount("/docs/", swagger_ui(schema_url="../openapi.json")) if not rate_limit_provider: rate_limit_provider = InMemoryLimitProvider( limit=100, timespan=300 ) public_app.mount( path="/login/", app=RateLimitingMiddleware( app=session_login( auth_table=self.auth_table, session_table=session_table, session_expiry=session_expiry, max_session_expiry=max_session_expiry, redirect_to=None, production=production, ), provider=rate_limit_provider, ), ) public_app.add_route( path="/logout/", route=session_logout(session_table=session_table), methods=["POST"], ) # We make the meta endpoint available without auth, because it contains # the site name. public_app.add_api_route( "/meta/", endpoint=self.get_meta, tags=["Meta"] # type: ignore ) # The translations are public, because we need them on the login page. public_app.add_api_route( "/translations/", endpoint=self.get_translation_list, # type: ignore methods=["GET"], tags=["Translations"], response_model=TranslationListResponse, ) public_app.add_api_route( "/translations/{language_code:str}/", endpoint=self.get_translation, # type: ignore methods=["GET"], tags=["Translations"], response_model=Translation, ) ####################################################################### self.router.add_route( path="/", endpoint=self.get_root, methods=["GET"] ) self.mount( path="/assets", app=StaticFiles(directory=os.path.join(ASSET_PATH, "assets")), ) auth_middleware = partial( AuthenticationMiddleware, backend=SessionsAuthBackend( auth_table=auth_table, session_table=session_table, admin_only=True, increase_expiry=increase_expiry, ), on_error=handle_auth_exception, ) self.mount(path="/api", app=auth_middleware(private_app)) self.mount(path="/public", app=public_app) async def get_root(self, request: Request) -> HTMLResponse: return HTMLResponse(self.template) ########################################################################### def _get_media_storage( self, table_name: str, column_name: str ) -> MediaStorage: """ Retrieve the ``MediaStorage`` for the given column. :raises HTTPException: If a matching ``MediaStorage`` can't be found. """ table_config = self.table_config_map.get(table_name) if not table_config: raise HTTPException(status_code=404, detail="No such table found.") media_columns = table_config.media_columns if media_columns is None: raise HTTPException( status_code=422, detail="No media columns are configured for this table.", ) try: column = table_config.table_class._meta.get_column_by_name( column_name ) except ValueError: raise HTTPException( status_code=404, detail="No such column found.", ) media_storage = media_columns.get(column) if not media_storage: raise HTTPException( status_code=422, detail="This column is not configured as a media_column.", ) return media_storage async def store_file( self, request: Request, table_name: str = Form(None), column_name: str = Form(None), file: UploadFile = File(...), ) -> StoreFileResponseModel: """ Stores in the file in the configured ``MediaStorage``, and returns a unique key for identifying that file. """ if self.read_only: raise HTTPException( status_code=405, detail="Running in read-only mode." ) media_storage = self._get_media_storage( table_name=table_name, column_name=column_name ) try: file_key = await media_storage.store_file( file_name=str(file.filename), file=file.file, user=request.user.user, ) except ValueError as exception: raise HTTPException(status_code=422, detail=str(exception)) return StoreFileResponseModel(file_key=file_key) async def generate_file_url( self, request: Request, model: GenerateFileURLRequestModel ) -> GenerateFileURLResponseModel: """ Returns a URL for accessing the given file. We don't use a GET for this endpoint, as using a GET param to pass the ``file_key`` is too restrictive on which characters can be used. """ if self.read_only: raise HTTPException( status_code=405, detail="Running in read-only mode." ) media_storage = self._get_media_storage( table_name=model.table_name, column_name=model.column_name ) file_url = await media_storage.generate_file_url( file_key=model.file_key, root_url=( f"./api/media-files/{model.table_name}" f"/{model.column_name}/" ), user=request.user.user, ) return GenerateFileURLResponseModel(file_url=file_url) ########################################################################### def get_user(self, request: Request) -> UserResponseModel: return UserResponseModel( username=request.user.display_name, user_id=str(request.user.user_id), ) ########################################################################### # Custom forms def get_forms(self) -> t.List[FormConfigResponseModel]: """ Returns a list of all forms registered with the admin. """ return [ FormConfigResponseModel( name=form.name, slug=form.slug, description=form.description ) for form in self.forms ] def get_single_form(self, form_slug: str) -> FormConfigResponseModel: """ Returns the FormConfig for the given form. """ form = self.form_config_map.get(form_slug, None) if form is None: raise HTTPException(status_code=404, detail="No such form found") else: return FormConfigResponseModel( name=form.name, slug=form.slug, description=form.description, ) def get_single_form_schema(self, form_slug: str) -> t.Dict[str, t.Any]: form_config = self.form_config_map.get(form_slug) if form_config is None: raise HTTPException(status_code=404, detail="No such form found") else: return form_config.pydantic_model.model_json_schema() async def post_single_form( self, request: Request, form_slug: str ) -> t.Any: """ Handles posting of custom forms. """ form_config = self.form_config_map.get(form_slug) data = await request.json() if form_config is None: raise HTTPException(status_code=404, detail="No such form found") try: model_instance = form_config.pydantic_model(**data) except ValidationError as exception: # We use 'detail' as it mirrors what FastAPI returns for Pydantic # errors, allowing us to use the same error display logic in the # front end. return JSONResponse( {"detail": json.loads(exception.json())}, status_code=422 ) try: endpoint = form_config.endpoint if inspect.iscoroutinefunction(endpoint): response = await endpoint( # type: ignore request, model_instance ) else: response = endpoint(request, model_instance) except ValueError as exception: return JSONResponse( {"custom_form_error": str(exception)}, status_code=422 ) message = ( response if isinstance(response, str) else "Successfully submitted" ) return JSONResponse({"custom_form_success": message}) ########################################################################### def get_meta(self) -> MetaResponseModel: return MetaResponseModel( piccolo_admin_version=PICCOLO_ADMIN_VERSION, site_name=self.site_name, ) ########################################################################### def get_sidebar_links(self) -> t.Dict[str, str]: """ Returns the custom links registered with the admin. """ return self.sidebar_links ########################################################################### def get_table_list(self) -> t.List[str]: """ Returns the list of table groups registered with the admin. """ return [i.table_class._meta.tablename for i in self.table_configs] def get_table_list_grouped(self) -> GroupedTableNamesResponseModel: """ Returns a list of all apps with tables registered with the admin, grouped using `menu_group`. """ response = GroupedTableNamesResponseModel() group_names = sorted( {i.menu_group for i in self.table_configs if i.menu_group} ) response.grouped = {i: [] for i in group_names} for table_config in self.table_configs: menu_group = table_config.menu_group if menu_group is None: response.ungrouped.append( table_config.table_class._meta.tablename ) else: response.grouped[menu_group].append( table_config.table_class._meta.tablename ) return response ########################################################################### def get_translation_list(self) -> TranslationListResponse: """ Return a list of language codes and names for each available translation. """ return TranslationListResponse( translations=[ TranslationListItem( language_code=translation.language_code, language_name=translation.language_name, ) for translation in self.translations_map.values() ], default_language_code=self.default_language_code, ) def get_translation(self, language_code: str = "en") -> Translation: """ Return a single language. The ``language_code`` is an IETF language code, for example 'en' for English. """ translation = self.translations_map.get(language_code.lower()) if translation is None: raise HTTPException( status_code=404, detail="Translation not found" ) return translation def get_all_tables( tables: t.Sequence[t.Type[Table]], ) -> t.Sequence[t.Type[Table]]: """ Fetch any related tables, and include them. """ output: t.List[t.Type[Table]] = [] def get_references(table: t.Type[Table]): references: t.List[t.Union[t.Type[Table], t.Any]] = [ i._foreign_key_meta.references for i in table._meta.foreign_key_columns ] for reference in references: table = ( reference.resolve() if isinstance(reference, LazyTableReference) else reference ) if table not in output: output.append(table) get_references(table) for table in tables: if table not in output: output.append(table) get_references(table) return output
[docs]def create_admin( tables: t.Sequence[t.Union[t.Type[Table], TableConfig]], forms: t.List[FormConfig] = [], auth_table: t.Optional[t.Type[BaseUser]] = None, session_table: t.Optional[t.Type[SessionsBase]] = None, session_expiry: timedelta = timedelta(hours=1), max_session_expiry: timedelta = timedelta(days=7), increase_expiry: t.Optional[timedelta] = timedelta(minutes=20), page_size: int = 15, read_only: bool = False, rate_limit_provider: t.Optional[RateLimitProvider] = None, production: bool = False, site_name: str = "Piccolo Admin", default_language_code: str = "auto", translations: t.Optional[t.List[Translation]] = None, auto_include_related: bool = True, allowed_hosts: t.Sequence[str] = [], debug: bool = False, sidebar_links: t.Dict[str, str] = {}, ): """ :param tables: Each of the tables will be added to the admin. :param forms: For each :class:`FormConfig <piccolo_admin.endpoints.FormConfig>` specified, a form will automatically be rendered in the user interface, accessible via the sidebar. :param auth_table: Either a :class:`BaseUser <piccolo.apps.user.tables.BaseUser>`, or ``BaseUser`` subclass table, which is used for fetching users. Defaults to ``BaseUser`` if none if specified. :param session_table: Either a :class:`SessionsBase <piccolo_api.session_auth.tables.SessionsBase>`, or ``SessionsBase`` subclass table, which is used for storing and querying session tokens. Defaults to ``SessionsBase`` if none if specified. :param session_expiry: How long a session is valid for. :param max_session_expiry: The maximum time a session is valid for, taking into account any refreshes using ``increase_expiry``. :param increase_expiry: If set, the ``session_expiry`` will be increased by this amount if it's close to expiry. :param page_size: The admin API paginates content - this sets the default number of results on each page. :param read_only: If ``True``, all non auth endpoints only respond to GET requests - the admin can still be viewed, and the data can be filtered. Useful for creating online demos. :param rate_limit_provider: Rate limiting middleware is used to protect the login endpoint against brute force attack. If not set, an :class:`InMemoryLimitProvider <piccolo_api.rate_limiting.middleware.InMemoryLimitProvider>` will be configured with reasonable defaults. :param production: If ``True``, the admin will enforce stronger security - for example, the cookies used will be secure, meaning they are only sent over HTTPS. :param site_name: Specify a different site name in the admin UI (default ``'Piccolo Admin'``). :param default_language_code: Specify the default language used in the admin UI. The value should be an `IETF language tag <https://en.wikipedia.org/wiki/IETF_language_tag>`_, for example ``'en'`` for English. To see available values see ``piccolo_admin/translations/data.py``. The UI will be automatically translated into this language. If a value of ``'auto'`` is specified, then we check the user's browser for the language they prefer, using the ``navigator.language`` JavaScript API. :param translations: Specify which translations are available. By default, we use every translation in ``piccolo_admin/translations/data.py``. Here's an example - if we know our users only speak English or Croatian, we can specify that only those translations are visible in the language selector in the UI:: from piccolo.translations.data import ENGLISH, CROATIAN create_admin( tables=[TableA, TableB], default_language_code='hr', translations=[ENGLISH, CROATIAN] ) You can also use this to provide your own translations, if there's a language we don't currently support (though please open a PR to add it!):: from piccolo.translations.models import Translation from piccolo.translations.data import ENGLISH MY_LANGUAGE = Translation( language_code='xx', language_name='My Language', translations={ 'Welcome': 'XXXXX', ... } ) create_admin( tables=[TableA, TableB], default_language_code='xx', translations=[ENGLISH, MY_LANGUAGE] ) :param auto_include_related: If a table has foreign keys to other tables, those tables will also be included in the admin by default, if not already specified. Otherwise the admin won't work as expected. :param allowed_hosts: This is used by the :class:`CSRFMiddleware <piccolo_api.csrf.middleware.CSRFMiddleware>` as an additional layer of protection when the admin is run under HTTPS. It must be a sequence of strings, such as ``['my_site.com']``. :param debug: If ``True``, debug mode is enabled. Any unhandled exceptions will return a stack trace, rather than a generic 500 error. Don't use this in production! :param sidebar_links: Custom links in the navigation sidebar. Example uses cases: * Providing a quick way to get to specific pages with pre-applied filters/sorting. * Linking to relative external websites. Here's a full example:: from piccolo_admin.endpoints import create_admin create_admin( tables=[Movie, Director], sidebar_links={ "Top Movies": "/admin/#/movie?__order=-box_office", "Google": "https://google.com" }, ) """ # noqa: E501 auth_table = auth_table or BaseUser session_table = session_table or SessionsBase if auto_include_related: table_config_map: t.Dict[t.Type[Table], t.Optional[TableConfig]] = {} for i in tables: if isinstance(i, TableConfig): table_config_map[i.table_class] = i else: table_config_map[i] = None all_table_classes = get_all_tables(tuple(table_config_map.keys())) all_table_classes_with_configs: t.List[ t.Union[t.Type[Table], TableConfig] ] = [] for i in all_table_classes: table_config = table_config_map.get(i) if table_config: all_table_classes_with_configs.append(table_config) else: all_table_classes_with_configs.append(i) tables = all_table_classes_with_configs return AdminRouter( *tables, forms=forms, auth_table=auth_table, session_table=session_table, session_expiry=session_expiry, max_session_expiry=max_session_expiry, increase_expiry=increase_expiry, page_size=page_size, read_only=read_only, rate_limit_provider=rate_limit_provider, production=production, site_name=site_name, default_language_code=default_language_code, translations=translations, allowed_hosts=allowed_hosts, debug=debug, sidebar_links=sidebar_links, )