flask_mongo_drf package

Subpackages

Submodules

flask_mongo_drf.mongo_decorators module

flask_mongo_drf.mongo_decorators.handle_api_errors(f)[source]

flask_mongo_drf.mongo_exceptions module

exception flask_mongo_drf.mongo_exceptions.APIException(message=None, code=None, payload=None)[source]

Bases: Exception

__init__(message=None, code=None, payload=None)[source]
message = 'Internal Server Error'
code = 500
exception flask_mongo_drf.mongo_exceptions.ValidationError(message=None, code=None, payload=None)[source]

Bases: APIException

code = 400
message = 'Validation failed'
exception flask_mongo_drf.mongo_exceptions.AuthenticationFailed(message=None, code=None, payload=None)[source]

Bases: APIException

code = 401
message = 'Authentication failed'
exception flask_mongo_drf.mongo_exceptions.PermissionDenied(message=None, code=None, payload=None)[source]

Bases: APIException

code = 403
message = 'Permission denied'
exception flask_mongo_drf.mongo_exceptions.NotFound(message=None, code=None, payload=None)[source]

Bases: APIException

code = 404
message = 'Resource not found'
exception flask_mongo_drf.mongo_exceptions.MongoDBConnectionError(message=None, code=None, payload=None)[source]

Bases: APIException

code = 500
message = 'MongoDB connection failed'

flask_mongo_drf.mongo_filters module

class flask_mongo_drf.mongo_filters.Filter(field_name: str | None = None, lookup_expr: str = 'exact', method: str | None = None, help_text: str | None = None)[source]

Bases: object

__init__(field_name: str | None = None, lookup_expr: str = 'exact', method: str | None = None, help_text: str | None = None)[source]
filter(value: Any) → Dict[str, Any][source]
class flask_mongo_drf.mongo_filters.CharFilter(field_name: str | None = None, lookup_expr: str = 'exact', method: str | None = None, help_text: str | None = None)[source]

Bases: Filter

class flask_mongo_drf.mongo_filters.NumberFilter(field_name: str | None = None, lookup_expr: str = 'exact', method: str | None = None, help_text: str | None = None)[source]

Bases: Filter

filter(value: Any) → Dict[str, Any][source]
class flask_mongo_drf.mongo_filters.BooleanFilter(field_name: str | None = None, lookup_expr: str = 'exact', method: str | None = None, help_text: str | None = None)[source]

Bases: Filter

filter(value: Any) → Dict[str, Any][source]
class flask_mongo_drf.mongo_filters.DateFilter(field_name: str | None = None, lookup_expr: str = 'exact', date_format: str = '%Y-%m-%d', **kwargs)[source]

Bases: Filter

__init__(field_name: str | None = None, lookup_expr: str = 'exact', date_format: str = '%Y-%m-%d', **kwargs)[source]
filter(value: Any) → Dict[str, Any][source]
class flask_mongo_drf.mongo_filters.ObjectIdFilter(field_name: str | None = None, lookup_expr: str = 'exact', method: str | None = None, help_text: str | None = None)[source]

Bases: Filter

filter(value: Any) → Dict[str, Any][source]
class flask_mongo_drf.mongo_filters.FilterSetMetaclass(name, bases, attrs)[source]

Bases: type

class flask_mongo_drf.mongo_filters.FilterSet(data: Dict[str, Any] | None = None, request=None)[source]

Bases: object

__init__(data: Dict[str, Any] | None = None, request=None)[source]
get_query_dict() → Dict[str, Any][source]

flask_mongo_drf.mongo_models module

class flask_mongo_drf.mongo_models.MongoBaseModel(collection: Collection)[source]

Bases: object

MongoDB 基础模型类。 通过依赖注入方式传入 PyMongo Collection 对象，与框架解耦。

__init__(collection: Collection)[source]
Parameters:

collection – PyMongo Collection 实例

count_documents(query: Dict[str, Any]) → int[source]

统计文档数量

find_all(query: Dict[str, Any], sort_by: str | None = None, sort_order: int = -1) → Iterator[Dict[str, Any]][source]

返回游标，避免一次性加载所有文档。 若需要排序，传入 sort_by 字段名和 sort_order（1 升序，-1 降序）。

find_paginated(query: Dict[str, Any], skip: int, limit: int, sort_by: str = '_id', sort_order: int = -1) → List[Dict[str, Any]][source]

分页查询，返回文档列表

insert_one(data: Dict[str, Any]) → InsertOneResult[source]

插入单条文档，自动添加 create_time 和 update_time

update_one_by_id(doc_id: str, data: Dict[str, Any]) → UpdateResult[source]

根据 _id 部分更新文档（PATCH）

replace_by_id(doc_id: str, new_doc: Dict[str, Any]) → UpdateResult[source]

根据 _id 全量替换文档（PUT）

delete_by_id(doc_id: str) → DeleteResult[source]

根据 _id 删除文档

find_one(query: Dict[str, Any]) → Dict[str, Any] | None[source]

查询单条文档

aggregate(pipeline: List[Dict[str, Any]])[source]

执行聚合管道，返回游标

flask_mongo_drf.mongo_paginations module

class flask_mongo_drf.mongo_paginations.MongoPagination(request_args, count, endpoint_name)[source]

Bases: object

page_size = 10
page_query_param = 'page'
page_size_query_param = 'page_size'
max_page_size = 100
__init__(request_args, count, endpoint_name)[source]
property skip
property limit
get_meta() → dict[source]

flask_mongo_drf.mongo_responses module

flask_mongo_drf.mongo_responses.custom_response(data=None, total=0, message='success', code=200, **kwargs)[source]

flask_mongo_drf.mongo_serializers module

class flask_mongo_drf.mongo_serializers.Field(required: bool = False, default: Any | None = None, allow_null: bool = False, read_only: bool = False, write_only: bool = False, validators: List[Callable] | None = None, help_text: str = '')[source]

Bases: object

字段基类

__init__(required: bool = False, default: Any | None = None, allow_null: bool = False, read_only: bool = False, write_only: bool = False, validators: List[Callable] | None = None, help_text: str = '')[source]
to_representation(value: Any) → Any[source]

序列化：将内部值转换为输出格式

to_internal_value(value: Any) → Any[source]

反序列化：将输入值转换为内部存储格式

validate(value: Any) → Any[source]

自定义校验，子类可重写

class flask_mongo_drf.mongo_serializers.CharField(max_length: int | None = None, min_length: int | None = None, **kwargs)[source]

Bases: Field

__init__(max_length: int | None = None, min_length: int | None = None, **kwargs)[source]
to_internal_value(value: Any) → str[source]

反序列化：将输入值转换为内部存储格式

class flask_mongo_drf.mongo_serializers.IntegerField(min_value: int | None = None, max_value: int | None = None, **kwargs)[source]

Bases: Field

__init__(min_value: int | None = None, max_value: int | None = None, **kwargs)[source]
to_internal_value(value: Any) → int[source]

反序列化：将输入值转换为内部存储格式

class flask_mongo_drf.mongo_serializers.FloatField(required: bool = False, default: Any | None = None, allow_null: bool = False, read_only: bool = False, write_only: bool = False, validators: List[Callable] | None = None, help_text: str = '')[source]

Bases: Field

to_internal_value(value: Any) → float[source]

反序列化：将输入值转换为内部存储格式

class flask_mongo_drf.mongo_serializers.BooleanField(required: bool = False, default: Any | None = None, allow_null: bool = False, read_only: bool = False, write_only: bool = False, validators: List[Callable] | None = None, help_text: str = '')[source]

Bases: Field

to_internal_value(value: Any) → bool[source]

反序列化：将输入值转换为内部存储格式

class flask_mongo_drf.mongo_serializers.DateTimeField(format: str = '%Y-%m-%d %H:%M:%S', **kwargs)[source]

Bases: Field

__init__(format: str = '%Y-%m-%d %H:%M:%S', **kwargs)[source]
to_representation(value: Any) → str[source]

序列化：将内部值转换为输出格式

to_internal_value(value: Any) → datetime[source]

反序列化：将输入值转换为内部存储格式

class flask_mongo_drf.mongo_serializers.ObjectIdField(required: bool = False, default: Any | None = None, allow_null: bool = False, read_only: bool = False, write_only: bool = False, validators: List[Callable] | None = None, help_text: str = '')[source]

Bases: Field

to_representation(value: Any) → str[source]

序列化：将内部值转换为输出格式

to_internal_value(value: Any) → ObjectId[source]

反序列化：将输入值转换为内部存储格式

class flask_mongo_drf.mongo_serializers.ListField(child: Field | None = None, **kwargs)[source]

Bases: Field

__init__(child: Field | None = None, **kwargs)[source]
to_representation(value: Any) → List[source]

序列化：将内部值转换为输出格式

to_internal_value(value: Any) → List[source]

反序列化：将输入值转换为内部存储格式

class flask_mongo_drf.mongo_serializers.DictField(required: bool = False, default: Any | None = None, allow_null: bool = False, read_only: bool = False, write_only: bool = False, validators: List[Callable] | None = None, help_text: str = '')[source]

Bases: Field

to_representation(value: Any) → Dict[source]

序列化：将内部值转换为输出格式

to_internal_value(value: Any) → Dict[source]

反序列化：将输入值转换为内部存储格式

class flask_mongo_drf.mongo_serializers.SerializerMetaclass(name, bases, attrs)[source]

Bases: type

class flask_mongo_drf.mongo_serializers.Serializer(instance=None, data=None, partial=False)[source]

Bases: object

__init__(instance=None, data=None, partial=False)[source]
property errors
property validated_data
is_valid(raise_exception=False)[source]
save(**kwargs)[source]
data()[source]
to_representation(obj)[source]
update(instance, validated_data)[source]
create(validated_data)[source]
class flask_mongo_drf.mongo_serializers.ModelSerializer(instance=None, data=None, partial=False, model=None)[source]

Bases: Serializer

__init__(instance=None, data=None, partial=False, model=None)[source]
to_representation(obj)[source]
create(validated_data)[source]
update(instance, validated_data)[source]

flask_mongo_drf.mongo_swagger module

flask_mongo_drf.mongo_swagger.generate_swagger_config(view_cls: Type) → Dict[str, Any][source]

flask_mongo_drf.mongo_viewsets module

class flask_mongo_drf.mongo_viewsets.MongoModelViewSet[source]

Bases: object

DRF 风格的 ModelViewSet，提供 list, create, retrieve, update, partial_update, destroy。

子类需要配置： - collection: PyMongo Collection 实例（必须） - serializer_class: 序列化器类（可选，默认自动生成） - filterset_class: 过滤器类（可选） - pagination_class: 分页器类（默认 MongoPagination） - default_sort_by / default_sort_order: 默认排序

model_class = None
collection = None
serializer_class: Type[Serializer] = None
filterset_class: Type[FilterSet] | None = None
pagination_class

alias of MongoPagination

default_sort_by: str = '_id'
default_sort_order: int = -1
__init__()[source]
get_serializer(instance=None, data=None, partial=False)[source]

获取序列化器实例

get_filter_dict() → Dict[str, Any][source]

从 request args 构建 MongoDB 过滤条件

get_pipeline(filter_dict: Dict[str, Any]) → List[Dict[str, Any]] | None[source]

子类可重写，返回聚合管道。 若返回非 None，则使用聚合管道查询（忽略排序和普通过滤）。

list()[source]

GET / 获取列表（支持过滤、分页、聚合）

create()[source]

POST / 创建新资源

retrieve(id)[source]

GET /<id> 获取单个资源

update(id)[source]

PUT /<id> 全量替换资源

partial_update(id)[source]

PATCH /<id> 部分更新资源

destroy(id)[source]

DELETE /<id> 删除资源

classmethod register_routes(blueprint: Blueprint, url_prefix: str, actions: List[str] | None = None)[source]

工业级路由注册，自动绑定 URL 规则和 Swagger 文档。

Module contents