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 ! lU 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.
"""
record = model.create(**element)
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
for k, value in element.items():
setattr(record, k, value)
record.save()
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 ! ) 8 apistar_crud/static/82f60bd0b94a1ed68b1e6e309ce2e8c3.svg
PK ! ) 8 apistar_crud/static/82f60bd0b94a1ed68b1e6e309ce2e8c3.svg
PK ! w 8 apistar_crud/static/8e3c7f5520f5ae906c6cf6d7f3ddcd19.eot 8 LP xPܵ &