版本 2.7<=, <=3.8
https://pypi.org/project/pymodm/
https://pymodm.readthedocs.io/en/stable/api/index.html#model-fields

github
https://github.com/mongodb/pymodm/

https://github.com/mongodb/pymodm/blob/master/test

example

models.py

  2. from pymodm.connection import connect
  3. connect("mongodb://root:example@192.168.204.129:27017/pintai?authSource=admin", alias="hwmg")
  5. import datetime as dt
  6. from pathlib import Path
  7. from pymodm.errors import ValidationError
  8. from pymongo.write_concern import WriteConcern
  9. from pymodm import MongoModel, fields
  12. def must_be_exists(value):
  13.     if not Path(value).exists():
  14.         raise ValidationError('指定的路径不存在')
  16.         class Magazine(MongoModel):
  17.             name = fields.CharField(required=True, min_length=5, max_length=20)
  18.             uptime = fields.DateTimeField(default=dt.datetime.utcnow)
  19.             # path = fields.CharField(required=True, validators=[must_be_exists])
  21.             class Meta:
  22.                 write_concern = WriteConcern(j =True)
  23.                 connection_alias = 'hwmg'
  24.         collection_name = 'MAGAZINE'

app.py

  1. from typing import Dict
  2. from models import Magazine
  3. from pathlib import Path
  4. from pymodm.errors import ValidationError
  8. def main():
  9.     # magazine = Magazine(name='安天安全周期第1期', path=r'/home/magazine/001.pdf')
  11.     # if magazine.is_valid():
  12.     #     magazine.save()
  14.     m1 = Magazine(name='安天')
  16.     # if m1.is_valid():
  17.     #     m1.save()
  18.     # m1.save()
  20.     try:
  21.         m1.full_clean()
  22.     except ValidationError as ve:
  23.         if isinstance(ve.message, dict):
  24.             # print(ve.message)
  25.             for k, v in ve.message.items():
  26.                 print(k,'=' ,v)
  28. if __name__ == "__main__":
  29.     main()

3

hwmg

  2. import pymongo
  3. import datetime as dt
  4. from datetime import timedelta
  5. from flask import request
  6. from pathlib import Path
  7. from app.models import Magazine
  8. from pymodm.errors import ValidationError
  10. from commonUtil.file_operation import filename_check, genDir, getSize
  11. # from module_server.resource_node.log_func import LOGGER
  13. from config.config import SECURITY_INFO_FILE_UPLOAD, MAX_ALLOW_UPLOAD_FILE_SIZE, ALLOW_UPLOAD_FILE_SUFFIX
  15. # 安全资讯文档上传存放路径　/home/hwmg/log/download/security_info
  16. genDir(SECURITY_INFO_FILE_UPLOAD)
  19. def get_sec_info_list(info_data):
  20.     """获取查询结果集成员, 生成列表
  22.     Args:
  23.         info_data: _description_
  24.     """
  25.     info_list = list()
  27.     for info in info_data:
  28.         item = {
  29.             "id": str(info.pk),
  30.             "name": info.name,
  31.             "uploadTime": info.uptime.timestamp(),
  32.             "url": f"download/security_info/{info.name}"
  33.         }
  34.         info_list.append(item)
  35.     return info_list
  38. def sec_info_get(json_data):
  39.     """获取安全资讯信息
  41.     Args:
  42.         json_data: _description_
  43.     """
  44.     page = int(json_data.get("page", 1))
  45.     limit = int(json_data.get("limit", 10))
  46.     name = json_data.get("name")
  47.     upload_time = json_data.get("uploadTime")
  48.     tt = dt.datetime.now()  # 查询时间段结束时间, to_time 默认当前
  49.     ft = tt - timedelta(days=7) # 查询时间段开始时间, from_time 默认7天前
  50.     payload = dict()
  52.     if upload_time:
  53.         from_ts, to_ts = upload_time.split(",")
  54.         ft = dt.datetime.fromtimestamp(int(from_ts))
  55.         tt = dt.datetime.fromtimestamp(int(to_ts))
  57.     find_qs = {"uptime": {"$gte": ft, "$lte": tt}} # 查询条件,时间段-上传时间
  59.     if name:
  60.         find_qs.update({"name": {"$regex": name}})
  62.     # 获取满足查询条件的所有结果， 实现排序
  63.     result = Magazine.objects.raw(find_qs).order_by([("uptime", pymongo.DESCENDING)])
  65.     if result.count() < 1:
  66.         payload['total'] = 0
  67.         payload['data'] = []
  68.     else:
  69.         payload['total'] = result.count()
  70.         payload['data'] = get_sec_info_list(result.limit(limit).skip((page - 1) * limit))
  72.     return dict(status=1, message="查询成功", payload = payload)
  75. def sec_info_post(json_dat):
  76.     """上传安全资讯
  78.     Args:
  79.         json_dat: _description_
  80.     """
  82.     sec_info_file = request.files.get("file")
  84.     if not sec_info_file:
  85.         return dict(status=-1, message="上传文件未找到")
  87.     sec_info_filename = filename_check(sec_info_file.filename)
  88.     sec_info_file_type = Path(sec_info_filename).suffix
  90.     if sec_info_file_type.lower() not in ALLOW_UPLOAD_FILE_SUFFIX:
  91.         return dict(status=-1, message="不支持的文件类型")
  93.     if sec_info_filename == sec_info_file_type:
  94.         return dict(status=-1, message="文件名称检查异常")
  96.     try:
  97.         Magazine.objects.get({"name":sec_info_filename})
  98.     except Magazine.DoesNotExist:
  100.         sec_info_file_save_path = SECURITY_INFO_FILE_UPLOAD / sec_info_filename
  101.         sec_info_file.save(str(sec_info_file_save_path))
  103.         if getSize(str(sec_info_file_save_path)) > 1024 * 1024 * MAX_ALLOW_UPLOAD_FILE_SIZE:
  104.             return dict(status=-1, message="文件大小超过限制 50M")
  106.         #　数据验证部分返回错误消息
  107.         try:
  108.             Magazine(name=sec_info_filename, path=str(sec_info_file_save_path)).save()
  109.             return dict(status=1, message="上传成功")
  110.         except ValidationError as ve:
  111.             for k, v in ve.message.items():
  112.                 if k == "name":
  113.                     return dict(status=-1, message=f"文件名长度不合法: {v}")
  114.                 else:
  115.                     return dict(status=-1, message=f"{v}")
  116.     else:
  117.         return dict(status=-1, message="文件已存在")