Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
from guillotina.interfaces import IPrincipalRoleManager
from guillotina.interfaces import IResourceSerializeToJson
from guillotina.registry import REGISTRY_DATA_KEY
from guillotina.response import ErrorResponse
from guillotina.response import HTTPConflict
from guillotina.response import HTTPNotFound
from guillotina.response import HTTPNotImplemented
from guillotina.response import HTTPPreconditionFailed
from guillotina.response import Response
from guillotina.utils import get_authenticated_user_id
from typing import Optional
import posixpath
@configure.service(
context=IDatabase,
method="GET",
permission="guillotina.GetContainers",
summary="Get list of containers",
responses={
"200": {
"description": "Get a list of containers",
"content": {
"application/json": {
"schema": {"properties": {"containers": {"type": "array", "items": {"type": "string"}}}}
}
},
}
},
)
class DefaultGET(Service):
from guillotina.response import ErrorResponse
from guillotina.response import HTTPNotFound
from guillotina.response import Response
from guillotina.schema import get_fields
from guillotina.utils import get_registry
from guillotina.utils import import_class
from guillotina.utils import resolve_dotted_name
_ = MessageFactory("guillotina")
_marker = object()
@configure.service(
context=IContainer,
method="GET",
permission="guillotina.ReadConfiguration",
name="@registry/{key}",
summary="Read container registry settings",
parameters=[{"in": "path", "name": "key", "required": True, "schema": {"type": "string"}}],
responses={
"200": {
"description": "Successfully registered interface",
"content": {
"application/json": {
"schema": {"type": "object", "properties": {"value": {"type": "object"}}}
}
},
}
},
from guillotina import configure
from guillotina.api.service import Service
from guillotina.auth.role import global_roles
from guillotina.interfaces import IContainer
@configure.service(
context=IContainer,
name="@available-roles",
permission="guillotina.ManageUsers",
summary="Get available roles on guillotina container",
method="GET",
)
class AvailableRoles(Service):
async def __call__(self):
roles = global_roles()
return roles
@configure.service(
context=IResource, method='GET',
permission='guillotina.SeePermissions', name='@sharing',
summary='Get sharing settings for this resource',
responses={
"200": {
"description": "All the sharing defined on this resource",
"schema": {
"$ref": "#/definitions/ResourceACL"
}
}
})
async def sharing_get(context, request):
roleperm = IRolePermissionMap(context)
prinperm = IPrincipalPermissionMap(context)
prinrole = IPrincipalRoleMap(context)
result = {
@configure.service(
context=IContainer, name="@cache-clear", method="POST", permission="guillotina.CacheManage"
)
async def clear(context, request):
utility = get_utility(ICacheUtility)
await utility.clear()
return {"success": True}
@configure.service(
context=IApplication,
method="GET",
permission="guillotina.GetContainers",
name="@apidefinition",
summary="Get API Definition",
description="Retrieves information on API configuration",
)
async def get_api_definition(context, request):
return app_settings["api_definition"]
'type': 'string'
}
}
}
}))
class TusHeadFile(UploadFile):
async def __call__(self):
# We need to get the upload as async IO and look for an adapter
# for the field to save there by chunks
adapter = getMultiAdapter(
(self.context, self.request, self.field), IFileManager)
return await adapter.tus_head()
@configure.service(
context=IResource, method='PATCH', permission='guillotina.ModifyContent',
name='@tusupload',
**_traversed_file_doc('TUS endpoint', parameters=[{
"name": "Upload-Offset",
"in": "headers",
"type": "integer",
"required": True
}, {
"name": "CONTENT-LENGTH",
"in": "headers",
"type": "integer",
"required": True
}], responses={
'204': {
'description': 'Successfully patched data',
'headers': {
responses={
"200": {
"description": "Successuflly deleted resource"
}
})
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 {}
if user is None:
raise HTTPUnauthorized(content={
'text': 'login failed'
})
jwt_token, data = authenticate_user(user.id)
await notify(UserLogin(user, jwt_token))
return {
'exp': data['exp'],
'token': jwt_token
}
@configure.service(
context=IContainer, method='POST',
permission='guillotina.AccessContent', name='@login-renew',
summary='Refresh to a new token')
@configure.service(
context=IApplication, method='POST',
permission='guillotina.AccessContent', name='@login-renew',
summary='Refresh to a new token')
class Refresh(Service):
token_timeout = 60 * 60 * 1
async def __call__(self):
user = get_authenticated_user()
data = {
'iat': datetime.utcnow(),
'exp': datetime.utcnow() + timedelta(seconds=self.token_timeout),
'id': user.id
@configure.service(
context=IContainer,
method="DELETE",
permission="guillotina.ManageAddons",
name="@addons",
summary="Uninstall an addon from container",
validate=True,
requestBody={
"required": True,
"content": {"application/json": {"schema": {"$ref": "#/components/schemas/Addon"}}},
},
)
async def uninstall(context, request):
data = await request.json()
id_to_uninstall = data.get("id", None)
return await uninstall_addon(context, request, id_to_uninstall)