python类List()的实例源码

def make_paginated_list_schema_class(cls, ns, item_schema):
        """
        Generate a schema class that represents a paginted list of items.

        """
        class PaginatedListSchema(Schema):
            __alias__ = "{}_list".format(ns.subject_name)
            items = fields.List(fields.Nested(item_schema), required=True)
            _links = fields.Raw()

        return PaginatedListSchema

def make_paginated_list_schema_class(cls, ns, item_schema):
        class PaginatedListSchema(Schema):
            __alias__ = "{}_list".format(ns.subject_name)

            offset = fields.Integer(required=True)
            limit = fields.Integer(required=True)
            count = fields.Integer(required=True)
            items = fields.List(fields.Nested(item_schema), required=True)
            _links = fields.Raw()

            @property
            def csv_column_order(self):
                return getattr(item_schema, "csv_column_order", None)

        return PaginatedListSchema

def build_parameter(field):
    """
    Build a parameter from a marshmallow field.

    See: https://github.com/marshmallow-code/apispec/blob/dev/apispec/ext/marshmallow/swagger.py#L81

    """
    if hasattr(field, SWAGGER_TYPE):
        parameter = resolve_tagged_field(field)
    elif getattr(field, "enum", None):
        parameter = resolve_enum_field(field)
    elif getattr(field, 'as_string', None):
        parameter = resolve_numeric_string_field(field)
    elif isinstance(field, TimestampField):
        parameter = resolve_timestamp_field(field)
    else:
        parameter = resolve_default_for_field(field)

    if field.metadata.get("description"):
        parameter["description"] = field.metadata["description"]

    if field.default:
        parameter["default"] = field.default

    # nested
    if isinstance(field, fields.Nested):
        parameter["$ref"] = "#/definitions/{}".format(type_name(name_for(field.schema)))

    # arrays
    if isinstance(field, fields.List):
        parameter["items"] = build_parameter(field.container)

    return parameter

def iter_schemas(marshmallow_schema):
    """
    Build zero or more JSON schemas for a marshmallow schema.

    Generates: name, schema pairs.

    """
    if not marshmallow_schema:
        return

    base_schema = build_schema(marshmallow_schema)
    base_schema_name = type_name(name_for(marshmallow_schema))
    yield base_schema_name, base_schema

    for name, field in iter_fields(marshmallow_schema):
        if isinstance(field, fields.Nested):
            nested_schema = build_schema(field.schema)
            nested_schema_name = type_name(name_for(field.schema))
            yield nested_schema_name, nested_schema
            for subname, subfield in iter_schemas(field.schema):
                yield subname, subfield
        if isinstance(field, fields.List) and isinstance(field.container, fields.Nested):
            nested_schema = build_schema(field.container.schema)
            nested_schema_name = type_name(name_for(field.container.schema))
            yield nested_schema_name, nested_schema
            for subname, subfield in iter_schemas(field.container.schema):
                yield subname, subfield

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 _postgres_array_factory(converter, data_type):
    return functools.partial(
        fields.List,
        converter._get_field_class_for_data_type(data_type.item_type),
    )

def property2field(self, prop, instance=True, **kwargs):
        field_class = self._get_field_class_for_property(prop)
        if not instance:
            return field_class
        field_kwargs = self._get_field_kwargs_for_property(prop)
        field_kwargs.update(kwargs)
        ret = field_class(**field_kwargs)
        if (
            hasattr(prop, 'direction') and
            self.DIRECTION_MAPPING[prop.direction.name] and
            prop.uselist is True
        ):
            ret = fields.List(ret, **kwargs)
        return ret

def _postgres_array_factory(converter, data_type):
    return functools.partial(
        fields.List,
        converter._get_field_class_for_data_type(data_type.item_type),
    )

def property2field(self, prop, instance=True, field_class=None, **kwargs):
        field_class = field_class or self._get_field_class_for_property(prop)
        if not instance:
            return field_class
        field_kwargs = self._get_field_kwargs_for_property(prop)
        field_kwargs.update(kwargs)
        ret = field_class(**field_kwargs)
        if (
            hasattr(prop, 'direction') and
            self.DIRECTION_MAPPING[prop.direction.name] and
            prop.uselist is True
        ):
            ret = fields.List(ret, **kwargs)
        return ret

def __init__(self, *args, **kwargs):
        kwargs.setdefault('default', [])
        kwargs.setdefault('missing', lambda: List(self.container))
        super().__init__(*args, **kwargs)

def _deserialize(self, value, attr, data):
        return List(self.container, super()._deserialize(value, attr, data))

def _deserialize_from_mongo(self, value):
        if value:
            return List(self.container, [self.container.deserialize_from_mongo(each)
                                         for each in value])
        else:
            return List(self.container)

def as_marshmallow_field(self, params=None, mongo_world=False):
        # Overwrite default `as_marshmallow_field` to handle deserialization
        # difference (`_id` vs `id`)
        kwargs = self._extract_marshmallow_field_params(mongo_world)
        if params:
            kwargs.update(params)
        return ma_fields.List(self.container.as_marshmallow_field(
            mongo_world=mongo_world), **kwargs)