How to use the guillotina.configure.adapter function in guillotina

To help you get started, we’ve selected a few guillotina examples, based on popular ways it is used in public projects.

Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.

github plone / guillotina / guillotina / json / deserialize_content.py View on Github external
from guillotina.interfaces import IResourceDeserializeFromJson
from guillotina.interfaces import RESERVED_ATTRS
from guillotina.schema import get_fields
from guillotina.schema.exceptions import ValidationError
from guillotina.utils import apply_coroutine
from guillotina.utils import get_security_policy
from zope.interface import Interface

import asyncio


logger = glogging.getLogger("guillotina")
_missing = object()


@configure.adapter(for_=(IResource, Interface), provides=IResourceDeserializeFromJson)
class DeserializeFromJson(object):
    def __init__(self, context, request):
        self.context = context
        self.request = request

        self.permission_cache = {}

    async def __call__(self, data, validate_all=False, ignore_errors=False, create=False):
        errors = []

        # do behavior first in case they modify context values
        for behavior_schema, behavior in await get_all_behaviors(self.context, load=False):
            dotted_name = behavior_schema.__identifier__
            if dotted_name not in data:
                # syntax {"namespace.IBehavior": {"foo": "bar"}}
                # we're not even patching this behavior if no iface found in payload
github plone / guillotina / guillotina / auth / participation.py View on Github external
from guillotina.auth import authenticate_request
from guillotina.auth.users import AnonymousUser
from guillotina.interfaces import Allow
from guillotina.interfaces import IParticipation
from guillotina.interfaces import IRequest


class AnonymousParticipation(object):

    def __init__(self, request):
        self.principal = AnonymousUser(request)
        self.principal._roles['guillotina.Anonymous'] = Allow
        self.interaction = None


@configure.adapter(for_=IRequest, provides=IParticipation)
class GuillotinaParticipation(object):
    principal = None

    def __init__(self, request):
        self.request = request

    async def __call__(self):
        # Cached user
        if not hasattr(self.request, '_cache_user'):
            user = await authenticate_request(self.request)
            if user is not None:
                self.request._cache_user = user
                self.principal = user
                if hasattr(user, 'roles') and 'guillotina.Authenticated' not in user.roles:
                    user.roles['guillotina.Authenticated'] = 1
        else:
github plone / guillotina / guillotina / db / partition.py View on Github external
from guillotina import configure
from guillotina.db.interfaces import IPartition
from guillotina.interfaces import IResource


@configure.adapter(
    for_=IResource,
    provides=IPartition)
class PartitionDataAdapter(object):

    def __init__(self, content):
        self.content = content

    def __call__(self):
        if hasattr(self.content, 'partition_id'):
            return self.content.partition_id
        else:
            return None
github plone / guillotina / guillotina / exc_resp.py View on Github external
def register_handler_factory(ExceptionKlass, factory):
    configure.adapter(for_=ExceptionKlass, provides=IErrorResponseException)(factory)
github plone / guillotina / guillotina / fields / dynamic.py View on Github external
return super().__call__(context, value)


@configure.adapter(for_=IDict, provides=IDynamicFieldOperation, name="update")
class DynamicDictUpdate(PatchDictUpdate):
    def __call__(self, context, value):
        if not isinstance(value, list):
            raise ValueDeserializationError(
                self.field, value, f"Invalid type patch data, must be list of updates"
            )
        for item in value:
            _validate_field(self.field, context, item)
        return super().__call__(context, value)


@configure.adapter(for_=IDict, provides=IDynamicFieldOperation, name="del")
class DynamicDictDel(PatchDictDel):
    """
    """
github plone / guillotina / guillotina / browser.py View on Github external
path.pop(1)
            path = "/".join(path)
        else:
            path = "/".join(get_physical_path(self.context))

        if container_url:
            return path
        elif relative:
            db = task_vars.db.get()
            return "/" + db.id + path
        else:
            db = task_vars.db.get()
            return get_url(self.request, db.id + path)


@configure.adapter(for_=IResource, provides=IAbsoluteURL)  # noqa: N801
class Absolute_URL_ObtainRequest(Absolute_URL):
    def __init__(self, context):
        request = get_current_request()
        super().__init__(context, request)
github plone / guillotina / guillotina / contrib / redis / dm.py View on Github external
from guillotina import configure
from guillotina.contrib.redis import get_driver
from guillotina.files.adapter import DBDataManager
from guillotina.interfaces import IExternalFileStorageManager
from guillotina.interfaces import IUploadDataManager
from guillotina.renderers import GuillotinaJSONEncoder
from guillotina.transactions import get_transaction

import json
import time


@configure.adapter(for_=IExternalFileStorageManager, provides=IUploadDataManager, name="redis")
class RedisFileDataManager(DBDataManager):

    _data = None
    _redis = None
    _loaded = False
    _ttl = 60 * 50 * 5  # 5 minutes should be plenty of time between activity

    async def load(self):
        # preload data
        if self._data is None:
            redis = await self.get_redis()
            key = self.get_key()
            data = await redis.get(key)
            if not data:
                self._data = {}
            else:
github plone / guillotina / guillotina / json / serialize_schema.py View on Github external
# -*- coding: utf-8 -*-
from guillotina import configure
from guillotina.component import get_multi_adapter
from guillotina.component import get_utility
from guillotina.component.interfaces import IFactory
from guillotina.interfaces import IBehavior
from guillotina.interfaces import IFactorySerializeToJson
from guillotina.interfaces import IRequest
from guillotina.interfaces import ISchemaFieldSerializeToJson
from guillotina.interfaces import ISchemaSerializeToJson
from guillotina.profile import profilable
from guillotina.schema import get_fields_in_order
from zope.interface import Interface


@configure.adapter(for_=(IFactory, IRequest), provides=IFactorySerializeToJson)
class SerializeFactoryToJson(object):
    def __init__(self, factory, request):
        self.factory = factory
        self.request = request

    @profilable
    async def __call__(self):
        factory = self.factory
        result = {
            "title": factory.type_name,
            "$schema": "http://json-schema.org/draft-04/schema#",
            "type": "object",
            "required": [],
            "definitions": {},
            "properties": {},
        }
github plone / guillotina / guillotina / db / cache / dummy.py View on Github external
from guillotina import configure
from guillotina.db.cache.base import BaseCache
from guillotina.db.interfaces import ITransaction
from guillotina.db.interfaces import ITransactionCache


@configure.adapter(for_=ITransaction, provides=ITransactionCache, name="dummy")
class DummyCache(BaseCache):
    async def get(self, **kwargs):
        return None

    async def set(self, value, **kwargs):
        pass

    async def clear(self):
        pass

    async def invalidate(self, ob):
        pass

    async def delete(self, key):
        pass
github plone / guillotina / guillotina / fields / annotation.py View on Github external
return value


@configure.value_serializer(BucketListValue)
def value_converter(value):
    if value is None:
        return
    return {"len": len(value), "buckets": len(value.annotations_metadata)}


@configure.value_serializer(BucketDictValue)
def value_dict_converter(value):
    return {"len": len(value), "buckets": len(value.buckets)}


@configure.adapter(for_=(Interface, IRequest, IBucketDictField), provides=IFieldValueRenderer)
class BucketDictFieldRenderer:
    def __init__(self, context, request, field):
        self.context = context
        self.request = request
        self.field = field

    async def __call__(self):
        """
        Iterate values bucket by bucket
        """
        val = self.field.get(self.field.context)
        if val is None:
            return {"values": {}, "total": 0, "cursor": None}
        bidx = 0
        if "cursor" in self.request.query:
            cursor = self.request.query["cursor"]