PK ! q apistar_crud/__init__.py__version__ = "0.2.3" __license__ = "GPLv3" __author__ = "José Antonio Perdiguero López" __email__ = "perdy.hh@gmail.com" __url__ = "https://github.com/PeRDy/apistar-crud" __description__ = "API Star tools to create CRUD resources." PK ! apistar_crud/admin.pyimport typing from apistar import App, Route, types, validators class Metadata(types.Type): resources = validators.Object(title="resources", description="Resource list") admin = validators.String(title="admin", description="Admin URL") schema = validators.String(title="schema", description="OpenAPI schema URL") class Admin: def __init__(self, *resources): self.resources = {resource.name: resource for resource in resources} def main(self, app: App, path: str = ""): """ Admin main page presenting a list of resources. """ return app.render_template("apistar_crud/admin.html") def metadata(self, app: App) -> Metadata: """ Admin metadata. """ return Metadata( { "resources": { resource.name: { "name": resource.name, "verbose_name": resource.verbose_name, "columns": resource.columns, "order": resource.order, } for resource in self.resources.values() }, "admin": app.reverse_url("admin:main"), "schema": app.reverse_url("serve_schema"), } ) @property def routes(self) -> typing.List[Route]: return [ Route("/", "GET", self.main, name="main", documented=False), Route("/{+path}", "GET", self.main, name="main-path", documented=False), ] @property def metadata_routes(self) -> typing.List[Route]: return [Route("/metadata/", "GET", self.metadata, name="metadata", documented=False)] PK ! ! apistar_crud/resource/__init__.pyPK ! apistar_crud/resource/base.pyimport re import typing from apistar import Route class BaseResource(type): METHODS = { "list": ("/", "GET"), # List resource collection "drop": ("/", "DELETE"), # Drop resource entire collection "create": ("/", "POST"), # Create a new element for this resource "retrieve": ("/{element_id}/", "GET"), # Retrieve an element of this resource "update": ("/{element_id}/", "PUT"), # Update an element of this resource "delete": ("/{element_id}/", "DELETE"), # Delete an element of this resource } AVAILABLE_METHODS = tuple(METHODS.keys()) DEFAULT_METHODS = ("create", "retrieve", "update", "delete", "list") def __new__(mcs, name, bases, namespace): try: resource_name = namespace.get("name", name.lower()) model = namespace["model"] input_type = namespace["input_type"] output_type = namespace["output_type"] except KeyError as e: raise AttributeError('{} needs to define attribute: "{}"'.format(name, e)) # Check resource name validity if re.match("[a-zA-Z][-_a-zA-Z]", resource_name) is None: raise AttributeError('Invalid resource name "{}"'.format(resource_name)) # Add valid name and verbose name namespace["name"] = resource_name namespace["verbose_name"] = namespace.get("verbose_name", namespace["name"]) # Default columns and order namespace["columns"] = namespace.get("columns", ["id"]) namespace["order"] = namespace.get("order", "id") methods = namespace.get("methods", mcs.DEFAULT_METHODS) # Create CRUD methods and routes mcs.add_methods(namespace, methods, model, input_type, output_type) mcs.add_routes(namespace, methods) return type(name, bases, namespace) @classmethod def add_routes(mcs, namespace: typing.Dict[str, typing.Any], methods: typing.Iterable[str]): class Routes: """Routes descriptor""" def __get__(self, instance, owner): return [Route(*mcs.METHODS[method], getattr(owner, method), name=method) for method in methods] namespace["routes"] = Routes() @classmethod def add_methods( mcs, namespace: typing.Dict[str, typing.Any], methods: typing.Iterable[str], model, input_type, output_type ): # Generate CRUD methods crud_ns = { k: v for m in methods for k, v in getattr(mcs, "add_{}".format(m))(model, input_type, output_type).items() } # Preserve already defined methods crud_ns.update({m: crud_ns[f"_{m}"] for m in methods if m not in namespace}) namespace.update(crud_ns) @classmethod def add_create(mcs, model, input_type, output_type) -> typing.Dict[str, typing.Any]: raise NotImplementedError @classmethod def add_retrieve(mcs, model, input_type, output_type) -> typing.Dict[str, typing.Any]: raise NotImplementedError @classmethod def add_update(mcs, model, input_type, output_type) -> typing.Dict[str, typing.Any]: raise NotImplementedError @classmethod def add_delete(mcs, model, input_type, output_type) -> typing.Dict[str, typing.Any]: raise NotImplementedError @classmethod def add_list(mcs, model, input_type, output_type) -> typing.Dict[str, typing.Any]: raise NotImplementedError @classmethod def add_drop(mcs, model, input_type, output_type) -> typing.Dict[str, typing.Any]: raise NotImplementedError PK ! apistar_crud/resource/peewee.pyimport operator import typing from functools import reduce from apistar import http, types, validators from apistar.exceptions import NotFound from apistar_pagination import PageNumberResponse from peewee import DoesNotExist from apistar_crud.resource.base import BaseResource class Resource(BaseResource): @classmethod def add_create(mcs, model, input_type, output_type) -> typing.Dict[str, typing.Any]: def create(cls, element: input_type) -> output_type: """ Create a new element for this resource. """ fields = dict(element) if fields.get("id") is None: del fields["id"] record = model.create(**fields) return http.JSONResponse(output_type(record), status_code=201) return {"_create": classmethod(create)} @classmethod def add_retrieve(mcs, model, input_type, output_type) -> typing.Dict[str, typing.Any]: def retrieve(cls, element_id: str) -> output_type: """ Retrieve an element of this resource. """ try: record = model.get_by_id(element_id) except DoesNotExist: raise NotFound return output_type(record) return {"_retrieve": classmethod(retrieve)} @classmethod def add_update(mcs, model, input_type, output_type) -> typing.Dict[str, typing.Any]: def update(cls, element_id: str, element: input_type) -> output_type: """ Update an element of this resource. """ try: record = model.get_by_id(element_id) except DoesNotExist: raise NotFound fields = dict(element) if fields.get("id") is None: del fields["id"] for k, value in fields.items(): setattr(record, k, value) if fields.get("id") is None: record.save() else: model.delete_by_id(element_id) record.save(force_insert=True) return http.JSONResponse(output_type(record), status_code=200) return {"_update": classmethod(update)} @classmethod def add_delete(mcs, model, input_type, output_type) -> typing.Dict[str, typing.Any]: def delete(cls, element_id: str) -> typing.Dict[str, typing.Any]: """ Delete an element of this resource. """ try: record = model.get_by_id(element_id) except DoesNotExist: raise NotFound record.delete_instance() return http.JSONResponse(None, status_code=204) return {"_delete": classmethod(delete)} @classmethod def add_list(mcs, model, input_type, output_type) -> typing.Dict[str, typing.Any]: def filter_(cls, **filters) -> typing.List[output_type]: """ Filtering over a resource collection. """ queryset = model.select() filters = {k: v for k, v in filters.items() if v} if filters: queryset = queryset.where(reduce(operator.and_, [(getattr(model, k) == v) for k, v in filters.items()])) return [output_type(record) for record in queryset] def list_(cls, page: http.QueryParam = None, page_size: http.QueryParam = None) -> typing.List[output_type]: """ List resource collection. """ return PageNumberResponse(page=page, page_size=page_size, content=cls._filter()) # noqa return {"_list": classmethod(list_), "_filter": classmethod(filter_)} @classmethod def add_drop(mcs, model, input_type, output_type) -> typing.Dict[str, typing.Any]: class DropOutput(types.Type): deleted = validators.Integer(title="deleted", description="Number of deleted elements", minimum=0) def drop(cls) -> DropOutput: """ Drop resource collection. """ num_records = model.delete().execute() return http.JSONResponse(DropOutput({"deleted": num_records}), status_code=204) return {"_drop": classmethod(drop)} PK ! b*Eq q # apistar_crud/resource/sqlalchemy.pyimport typing from apistar import http, types, validators from apistar.exceptions import NotFound from apistar_pagination import PageNumberResponse from sqlalchemy.orm import Session from apistar_crud.resource.base import BaseResource class Resource(BaseResource): @classmethod def add_create(mcs, model, input_type, output_type) -> typing.Dict[str, typing.Any]: def create(cls, session: Session, element: input_type) -> output_type: """ Create a new element for this resource. """ record = model(**element) session.add(record) session.flush() return http.JSONResponse(output_type(record), status_code=201) return {"_create": classmethod(create)} @classmethod def add_retrieve(mcs, model, input_type, output_type) -> typing.Dict[str, typing.Any]: def retrieve(cls, session: Session, element_id: str) -> output_type: """ Retrieve an element of this resource. """ record = session.query(model).get(element_id) if record is None: raise NotFound return output_type(record) return {"_retrieve": classmethod(retrieve)} @classmethod def add_update(mcs, model, input_type, output_type) -> typing.Dict[str, typing.Any]: def update(cls, session: Session, element_id: str, element: input_type) -> output_type: """ Update an element of this resource. """ record = session.query(model).get(element_id) if record is None: raise NotFound for k, value in element.items(): setattr(record, k, value) return http.JSONResponse(output_type(record), status_code=200) return {"_update": classmethod(update)} @classmethod def add_delete(mcs, model, input_type, output_type) -> typing.Dict[str, typing.Any]: def delete(cls, session: Session, element_id: str) -> typing.Dict[str, typing.Any]: """ Delete an element of this resource. """ if session.query(model).filter_by(id=element_id).count() == 0: raise NotFound session.query(model).filter_by(id=element_id).delete() return http.JSONResponse(None, status_code=204) return {"_delete": classmethod(delete)} @classmethod def add_list(mcs, model, input_type, output_type) -> typing.Dict[str, typing.Any]: def filter_(cls, session: Session, **filters) -> typing.List[output_type]: """ Filter resource collection. """ filters = {k: v for k, v in filters.items() if v} if filters: queryset = session.query(model).filter_by(**filters) else: queryset = session.query(model).all() return [output_type(record) for record in queryset] def list_(cls, page: http.QueryParam = None, page_size: http.QueryParam = None) -> typing.List[output_type]: """ List resource collection. """ return PageNumberResponse(page=page, page_size=page_size, content=cls._filter()) # noqa return {"_list": classmethod(list_), "_filter": classmethod(filter_)} @classmethod def add_drop(mcs, model, input_type, output_type) -> typing.Dict[str, typing.Any]: class DropOutput(types.Type): deleted = validators.Integer(title="deleted", description="Number of deleted elements", minimum=0) def drop(cls, session: Session) -> DropOutput: """ Drop resource collection. """ num_records = session.query(model).count() session.query(model).delete() return http.JSONResponse(DropOutput({"deleted": num_records}), status_code=204) return {"_drop": classmethod(drop)} PK ! 3CrO O apistar_crud/routes.pyimport typing from collections import namedtuple from apistar import Include from apistar_crud.admin import Admin __all__ = ["routes"] RouteOptions = namedtuple("RouteOptions", ("url", "admin")) class Routes: def __init__(self): self.resources = {} def register(self, resource, url: str = None, admin: bool = True): """ Register a resource. :param resource: Resource. :param url: Route url. :param admin: True if should be added to admin site. """ url = url or "/{}".format(resource.name) self.resources[resource] = RouteOptions(url, admin) def routes(self, admin="/admin") -> typing.List[Include]: """ Generate the list of routes for all resources and admin views. :param admin: Admin path, disabled if None. :return: List of routes. """ r = [Include(opts.url, r.name, r.routes) for r, opts in self.resources.items()] private_routes = [] if admin: a = Admin(*[r for r, opts in self.resources.items() if opts.admin]) r.append(Include(admin, "admin", a.routes, documented=False)) private_routes += a.metadata_routes if private_routes: r.append(Include("/_crud", "crud", private_routes, documented=False)) return r routes = Routes() PK ! Ԝ Ԝ : apistar_crud/static/0ab54153eeeca0ce03978cc463b257f7.woff2wOF2 8 TV : X[6$Pj z[Uq&{[" 0RCqXt ▻pTd4[ QUUO$ S(мq۲.XgEl?+)( bBâ9R<6̉9"