Python marshmallow.fields 模块，Field() 实例源码

def resolve_default_for_field(field):
    """
    Field defaults should be the fallback.

    """
    try:
        field_type, field_format = FIELD_MAPPINGS[type(field)]
        if field_format:
            return dict(
                type=field_type,
                format=field_format,
            )
        elif field_type:
            return dict(
                type=field_type,
            )
        else:
            return dict()
    except KeyError:
        logger.exception("No mapped swagger type for marshmallow field: {}".format(
            field,
        ))
        raise

def serialize(self, attr_name, field_symbol,
                  assignment_template, field_obj):
        # type: (str, str, str, fields.Field) -> IndentedString
        """Generates the code to pull a field off of an object into the result.

        :param attr_name: The name of the attribute being accessed/
        :param field_symbol: The symbol to use when accessing the field.  Should
            be generated via field_symbol_name.
        :param assignment_template: A string template to use when generating
            code.  The assignment template is passed into the serializer and
            has a single possitional placeholder for string formatting.  An
            example of a value that may be passed into assignment_template is:
            `res['some_field'] = {0}`
        :param field_obj: The instance of the Marshmallow field being
            serialized.
        :return: The code to pull a field off of the object passed in.
        """
        pass  # pragma: no cover

def inline(self, field, context):
        # type: (fields.Field, JitContext) -> Optional[str]
        """Generates a template for inlining boolean serialization.

        For example, generates:

        (
            (value in __some_field_truthy) or
            (False if value in __some_field_falsy else bool(value))
        )

        This is somewhat fragile but it tracks what Marshmallow does.
        """
        if is_overridden(field._serialize, fields.Boolean._serialize):
            return None
        truthy_symbol = '__{0}_truthy'.format(field.name)
        falsy_symbol = '__{0}_falsy'.format(field.name)
        context.namespace[truthy_symbol] = field.truthy
        context.namespace[falsy_symbol] = field.falsy
        result = ('(({0} in ' + truthy_symbol +
                  ') or (False if {0} in ' + falsy_symbol +
                  ' else dict()["error"]))')
        return result + ' if {0} is not None else None'

def inline(self, field, context):
        # type: (fields.Field, JitContext) -> Optional[str]
        """Generates a template for inlining string serialization.

        For example, generates "float(value) if value is not None else None"
        to serialize a float.  If `field.as_string` is `True` the result will
        be coerced to a string if not None.
        """
        if (is_overridden(field._validated, fields.Number._validated) or
                is_overridden(field._serialize, fields.Number._serialize)):
            return None
        result = field.num_type.__name__ + '({0})'
        if field.as_string and context.is_serializing:
            result = 'str({0})'.format(result)
        if field.allow_none is True:
            # Only emit the Null checking code if nulls are allowed.  If they
            # aren't allowed casting `None` to an integer will throw and the
            # slow path will take over.
            result += ' if {0} is not None else None'
        return result

def _collect_schema_attrs(nmspc):
    """
    Split dict between schema fields and non-fields elements and retrieve
    marshmallow tags if any.
    """
    schema_fields = {}
    schema_non_fields = {}
    doc_nmspc = {}
    for key, item in nmspc.items():
        if hasattr(item, '__marshmallow_tags__'):
            # Decorated special functions (e.g. `post_load`)
            schema_non_fields[key] = item
        elif isinstance(item, Field):
            # Given the fields provided by the template are going to be
            # customized in the implementation, we copy them to avoid
            # overwriting if two implementations are created
            schema_fields[key] = copy(item)
        else:
            doc_nmspc[key] = item
    return doc_nmspc, schema_fields, schema_non_fields

def on_need_add_id_field(bases, fields):
    """
    If the given fields make no reference to `_id`, add an `id` field
    (type ObjectId, dump_only=True, attribute=`_id`) to handle it
    """

    def find_id_field(fields):
        for name, field in fields.items():
            # Skip fake fields present in schema (e.g. `post_load` decorated function)
            if not isinstance(field, Field):
                continue
            if (name == '_id' and not field.attribute) or field.attribute == '_id':
                return name, field

    # Search among parents for the id field
    for base in bases:
        schema = base()
        if find_id_field(schema.fields):
            return

    # Search amongo our own fields
    if not find_id_field(fields):
        # No id field found, add a default one
        from .fields import ObjectIdField
        fields['id'] = ObjectIdField(attribute='_id', dump_only=True)

def dump_schema(schema_obj):
    json_schema = {
        "type": "object",
        "properties": {},
        "required": [],
    }
    mapping = {v: k for k, v in schema_obj.TYPE_MAPPING.items()}
    mapping[fields.Email] = text_type
    mapping[fields.Dict] = dict
    mapping[fields.List] = list
    mapping[fields.Url] = text_type
    mapping[fields.LocalDateTime] = datetime.datetime

    for field_name, field in sorted(schema_obj.fields.items()):
        schema = None

        if field.__class__ in mapping:
            pytype = mapping[field.__class__]
            schema = _from_python_type(field, pytype)
        elif isinstance(field, fields.Nested):
            schema = _from_nested_schema(field)
        elif issubclass(field.__class__, fields.Field):
            for cls in mapping.keys():
                if issubclass(field.__class__, cls):
                    pytype = mapping[cls]
                    schema = _from_python_type(field, pytype)
                    break

        if schema is None:
            raise ValueError('unsupported field type %s' % field)

        field_name = field.dump_to or field.name
        json_schema['properties'][field_name] = schema
        if field.required:
            json_schema['required'].append(field.name)
    return json_schema

def serialize(self, attr_name, field_symbol,
                  assignment_template, field_obj):
        # type: (str, str, str, fields.Field) -> IndentedString
        return IndentedString(assignment_template.format(attr_str(attr_name)))

def serialize(self, attr_name, field_symbol,
                  assignment_template, field_obj):
        # type: (str, str, str, fields.Field) -> IndentedString
        body = IndentedString()
        if self.context.is_serializing:
            default_str = 'default'
            default_value = field_obj.default
        else:
            default_str = 'missing'
            default_value = field_obj.missing
            if field_obj.required:
                body += assignment_template.format('obj["{attr_name}"]'.format(
                    attr_name=attr_name))
                return body
        if default_value == missing:
            body += 'if "{attr_name}" in obj:'.format(attr_name=attr_name)
            with body.indent():
                body += assignment_template.format('obj["{attr_name}"]'.format(
                    attr_name=attr_name))
        else:
            if callable(default_value):
                default_str += '()'

            body += assignment_template.format(
                'obj.get("{attr_name}", {field_symbol}__{default_str})'.format(
                    attr_name=attr_name, field_symbol=field_symbol,
                    default_str=default_str))
        return body

def serialize(self, attr_name, field_symbol,
                  assignment_template, field_obj):
        # type: (str, str, str, fields.Field) -> IndentedString
        body = IndentedString()
        body += 'try:'
        with body.indent():
            body += 'value = obj["{attr_name}"]'.format(attr_name=attr_name)
        body += 'except (KeyError, AttributeError, IndexError, TypeError):'
        with body.indent():
            body += 'value = {attr_str}'.format(attr_str=attr_str(attr_name))
        body += assignment_template.format('value')
        return body

def inline(self, field, context):
        # type: (fields.Field, JitContext) -> Optional[str]
        pass  # pragma: no cover

def _should_skip_field(field_name, field_obj, context):
    # type: (str, fields.Field, JitContext) -> bool
    if (getattr(field_obj, 'load_only', False) and
            context.is_serializing):
        return True
    if (getattr(field_obj, 'dump_only', False) and
            not context.is_serializing):
        return True
    if context.only and field_name not in context.only:
        return True
    if context.exclude and field_name in context.exclude:
        return True
    return False

def _get_attr_and_destination(context, field_name, field_obj):
    # type: (JitContext, str, fields.Field) -> Tuple[str, str]
    # The name of the attribute to pull off the incoming object
    attr_name = field_name
    # The destination of the field in the result dictionary.
    destination = field_name
    if context.is_serializing:
        destination = field_obj.dump_to or field_name
    if field_obj.attribute:
        if context.is_serializing:
            attr_name = field_obj.attribute
        else:
            destination = field_obj.attribute
    return attr_name, destination

def inliner_for_field(context, field_obj):
    # type: (JitContext, fields.Field) -> Optional[str]
    if context.use_inliners:
        inliner = None
        for field_type, inliner_class in iteritems(INLINERS):
            if isinstance(field_obj, field_type):
                inliner = inliner_class.inline(field_obj, context)
                if inliner:
                    break
        return inliner
    return None

def _is_field(value):
    return (
        isinstance(value, type) and
        issubclass(value, fields.Field)
    )

def __init__(self, endpoint, **kwargs):
        self.endpoint = endpoint
        self.params = kwargs
        fields.Field.__init__(self, **kwargs)

def __init__(self, schema, **kwargs):
        self.schema = schema
        fields.Field.__init__(self, **kwargs)

def _is_field(value):
    return (
        isinstance(value, type) and
        issubclass(value, fields.Field)
    )

def get_declared_fields(mcs, klass, cls_fields, inherited_fields, dict_cls):
        """Returns a dictionary of field_name => `Field` pairs declard on the class.
        This is exposed mainly so that plugins can add additional fields, e.g. fields
        computed from class Meta options.

        :param type klass: The class object.
        :param dict cls_fields: The fields declared on the class, including those added
            by the ``include`` class Meta option.
        :param dict inherited_fileds: Inherited fields.
        :param type dict_class: Either `dict` or `OrderedDict`, depending on the whether
            the user specified `ordered=True`.
        """
        return dict_cls(inherited_fields + cls_fields)

    # NOTE: self is the class object

def __init__(self, extra=None, only=(), exclude=(), prefix='', strict=None,
                 many=False, context=None, load_only=(), dump_only=(),
                 partial=False):
        # copy declared fields from metaclass
        self.declared_fields = copy.deepcopy(self._declared_fields)
        self.many = many
        self.only = only
        self.exclude = exclude
        self.prefix = prefix
        self.strict = strict if strict is not None else self.opts.strict
        self.ordered = self.opts.ordered
        self.load_only = set(load_only) or set(self.opts.load_only)
        self.dump_only = set(dump_only) or set(self.opts.dump_only)
        self.partial = partial
        #: Dictionary mapping field_names -> :class:`Field` objects
        self.fields = self.dict_class()
        #: Callable marshalling object
        self._marshal = marshalling.Marshaller(
            prefix=self.prefix
        )
        #: Callable unmarshalling object
        self._unmarshal = marshalling.Unmarshaller()
        if extra:
            warnings.warn(
                'The `extra` argument is deprecated. Use a post_dump '
                'method to add additional data instead.',
                DeprecationWarning
            )
        self.extra = extra
        self.context = context or {}
        self._normalize_nested_options()
        self._types_seen = set()
        self._update_fields(many=many)

def test_add_parser_from_field(self, set_env, env):
        class MyURL(fields.Field):
            def _deserialize(self, value, *args, **kwargs):
                return 'https://' + value

        env.add_parser_from_field('url', MyURL)

        set_env({'URL': 'test.test/'})
        assert env.url('URL') == 'https://test.test/'

        with pytest.raises(environs.EnvError) as excinfo:
            env.url('NOT_SET')
        assert excinfo.value.args[0] == 'Environment variable "NOT_SET" not set'

def __filter_fields(self, field_names, obj, many=False):
        """Return only those field_name:field_obj pairs specified by
        ``field_names``.

        :param set field_names: Field names to include in the final
            return dictionary.
        :returns: An dict of field_name:field_obj pairs.
        """
        if obj and many:
            try:  # Homogeneous collection
                # Prefer getitem over iter to prevent breaking serialization
                # of objects for which iter will modify position in the collection
                # e.g. Pymongo cursors
                if hasattr(obj, '__getitem__') and callable(getattr(obj, '__getitem__')):
                    obj_prototype = obj[0]
                else:
                    obj_prototype = next(iter(obj))
            except (StopIteration, IndexError):  # Nothing to serialize
                return self.declared_fields
            obj = obj_prototype
        ret = self.dict_class()
        for key in field_names:
            if key in self.declared_fields:
                ret[key] = self.declared_fields[key]
            else:  # Implicit field creation (class Meta 'fields' or 'additional')
                if obj:
                    attribute_type = None
                    try:
                        if isinstance(obj, Mapping):
                            attribute_type = type(obj[key])
                        else:
                            attribute_type = type(getattr(obj, key))
                    except (AttributeError, KeyError) as err:
                        err_type = type(err)
                        raise err_type(
                            '"{0}" is not a valid field for {1}.'.format(key, obj))
                    field_obj = self.TYPE_MAPPING.get(attribute_type, fields.Field)()
                else:  # Object is None
                    field_obj = fields.Field()
                # map key -> field (default to Raw)
                ret[key] = field_obj
        return ret