Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
async def default_head(context, request):
return {}
@configure.service(
context=IResource,
method="GET",
permission="guillotina.ViewContent",
summary="Retrieves serialization of resource",
responses=get_content_json_schema_responses,
parameters=[
{"name": "include", "in": "query", "required": True, "schema": {"type": "string"}},
{"name": "omit", "in": "query", "required": True, "schema": {"type": "string"}},
],
)
class DefaultGET(Service):
@profilable
async def __call__(self):
serializer = get_multi_adapter((self.context, self.request), IResourceSerializeToJson)
include = omit = []
if self.request.query.get("include"):
include = self.request.query.get("include").split(",")
if self.request.query.get("omit"):
omit = self.request.query.get("omit").split(",")
try:
result = await serializer(include=include, omit=omit)
except TypeError:
result = await serializer()
await notify(ObjectVisitedEvent(self.context))
return result
requestBody={
"required": True,
"content": {
"application/json": {
"schema": {
"type": "object",
"title": "Registry",
"properties": {"interface": {"type": "string"}, "initial_values": {"type": "object"}},
"required": ["interface"],
}
}
},
},
responses={"200": {"description": "Successfully registered interface"}},
)
class Register(Service):
"""Register an Interface on the Registry."""
async def __call__(self):
""" data input : { 'interface': 'INTERFACE' }"""
registry = await get_registry()
if registry is None:
return ErrorResponse("BadRequest", _("Not in a container request"), status=412)
data = await self.request.json()
interface = data.get("interface", None)
initial_values = data.get("initial_values", {})
if interface is None:
return ErrorResponse("InvalidRequest", "Non existent Interface", status=412)
iObject = import_class(interface)
registry.register_interface(iObject)
}
})
class DefaultDELETE(Service):
async def __call__(self):
content_id = self.context.id
parent = self.context.__parent__
await notify(BeforeObjectRemovedEvent(self.context, parent, content_id))
self.context._p_jar.delete(self.context)
await notify(ObjectRemovedEvent(self.context, parent, content_id))
@configure.service(
context=IResource, method='OPTIONS', permission='guillotina.AccessPreflight',
summary='Get CORS information for resource')
class DefaultOPTIONS(Service):
"""Preflight view for Cors support on DX content."""
def getRequestMethod(self): # noqa
"""Get the requested method."""
return self.request.headers.get(
'Access-Control-Request-Method', None)
async def preflight(self):
"""We need to check if there is cors enabled and is valid."""
headers = {}
if not app_settings['cors']:
return {}
origin = self.request.headers.get('Origin', None)
if not origin:
from guillotina.utils import get_authenticated_user
from guillotina.utils import navigate_to
from zope.interface import alsoProvides
import typing
@configure.service(
context=IContainer,
name="@user_info",
permission="guillotina.AccessContent",
method="GET",
summary="Get info about authenticated user",
allow_access=True,
)
class Info(Service):
async def __call__(self):
user = get_authenticated_user()
serializer = queryMultiAdapter((user, self.request), IResourceSerializeToJson)
if serializer:
data = await serializer()
else:
data = {}
data.update({"id": user.id, "roles": user.roles, "groups": getattr(user, "groups", [])})
return data
class BaseUser(Service):
async def get_user(self) -> User:
user_id: str = self.request.matchdict["user"]
try:
user = await navigate_to(self.context, "users/{}".format(user_id))
klass = func
original = klass.__call__
call_original = klass._call_original
else:
if not _has_parameters(func):
raise ServiceConfigurationError(
f"Service configuration must accept 2 required parameters: {func}\n"
f"{pformat(self.config)}"
)
if not asyncio.iscoroutinefunction(func):
raise ServiceConfigurationError(
f"Service function must be async: {func}\n" f"{pformat(self.config)}"
)
from guillotina.api.service import Service
klass = self.config.get("base", Service)
original = staticmethod(func)
call_original = klass._call_original_func
call = call_original
if self.config.get("validate", False):
call = klass._call_validate
# create new class with customizations
klass = type(
func.__name__,
(klass,),
{
**dict(klass.__dict__),
**{
"__module__": func.__module__,
"__allow_access__": self.config.get(
"allow_access", getattr(klass, "__allow_access__", False)
return result
@configure.service(
context=IResource,
method="POST",
permission="guillotina.ChangePermissions",
name="@sharing",
summary="Change permissions for a resource",
requestBody={
"required": True,
"content": {"application/json": {"schema": {"$ref": "#/components/schemas/Permissions"}}},
},
responses={"200": {"description": "Successfully changed permission"}},
)
class SharingPOST(Service):
async def __call__(self, changed=False):
"""Change permissions"""
context = self.context
request = self.request
data = await request.json()
if (
"prinrole" not in data
and "roleperm" not in data
and "prinperm" not in data
and "perminhe" not in data
):
raise PreconditionFailed(self.context, "prinrole or roleperm or prinperm missing")
return await apply_sharing(context, data)
@configure.service(
parameters=[{
"name": "body",
"in": "body",
"schema": {
"$ref": "#/definitions/AddableResource"
}
}],
responses={
"200": {
"description": "Resource data",
"schema": {
"$ref": "#/definitions/ResourceFolder"
}
}
})
class DefaultPOST(Service):
async def __call__(self):
"""To create a content."""
data = await self.get_data()
type_ = data.get('@type', None)
id_ = data.get('id', None)
behaviors = data.get('@behaviors', None)
if '__acl__' in data:
# we don't allow to change the permisions on this patch
del data['__acl__']
if not type_:
return ErrorResponse(
'RequiredParam',
_("Property '@type' is required"))
from urllib.parse import urlparse
from zope.interface import Interface
import copy
import json
import os
import pkg_resources
here = os.path.dirname(os.path.realpath(__file__))
@configure.service(
method="GET", context=Interface, name="@swagger", permission="guillotina.swagger.View", ignore=True
)
class SwaggerDefinitionService(Service):
__allow_access__ = True
def get_data(self, data):
if callable(data):
data = data(self.context)
return data
def load_swagger_info(self, api_def, path, method, tags, service_def):
path = path.rstrip("/")
if path not in api_def:
api_def[path or "/"] = {}
desc = self.get_data(service_def.get("description", ""))
swagger_conf = service_def.get("swagger", {})
if swagger_conf.get("display_permission", True):
if desc:
desc += f" 〜 permission: {service_def['permission']}"
@configure.service(
context=IContainer,
method="GET",
permission="guillotina.AccessContent",
name="@types/{type_name}",
summary="Read information on available types",
parameters=[{"in": "path", "name": "type_name", "required": True, "schema": {"type": "string"}}],
responses={
"200": {
"description": "Result results on types",
"content": {"application/json": {"schema": {"properties": {}}}},
}
},
)
class Read(Service):
async def prepare(self):
type_name = self.request.matchdict["type_name"]
self.value = query_utility(IResourceFactory, name=type_name)
if self.value is None:
raise HTTPNotFound(content={"reason": f"Could not find type {type_name}", "type_name": type_name})
async def __call__(self):
serializer = get_multi_adapter((self.value, self.request), IFactorySerializeToJson)
return await serializer()
)
class ListGroups(ListGroupsOrUsersService):
type_name: str = "Group"
_desired_keys: typing.List[str] = ["groupname", "id", "title", "roles", "users", "@name"]
async def process_catalog_obj(self, obj) -> dict:
return {
"@name": obj.get("@name"),
"id": obj.get("id"),
"title": obj.get("group_name"),
"users": obj.get("group_users") or [],
"roles": obj.get("group_user_roles") or [],
}
class BaseGroup(Service):
async def get_group(self) -> Group:
group_id: str = self.request.matchdict["group"]
group: typing.Optional[Group] = await navigate_to(self.context, "groups/{}".format(group_id))
if not group:
raise HTTPNotFound(content={"reason": f"Group {group} not found"})
return group
@configure.service(
for_=IContainer,
methods="GET",
name="@groups/{group}",
permission="guillotina.ManageUsers",
responses={
"200": {
"description": "Group",