MongoDB

数据自动入Mongo库使用须知

  • 使用MongoDb存储数据，需要使用MongoPipeline

示例:

  1. import feapder
  2. from feapder import Item
  5. class TestMongo(feapder.AirSpider):
  6.     __custom_setting__ = dict(
  7.         ITEM_PIPELINES=["feapder.pipelines.mongo_pipeline.MongoPipeline"],
  8.         MONGO_IP="localhost",
  9.         MONGO_PORT=27017,
  10.         MONGO_DB="feapder",
  11.         MONGO_USER_NAME="",
  12.         MONGO_USER_PASS="",
  13.     )
  15.     def start_requests(self):
  16.         yield feapder.Request("https://www.baidu.com")
  18.     def parse(self, request, response):
  19.         title = response.xpath("//title/text()").extract_first()  # 取标题
  20.         item = Item()  # 声明一个item
  21.         item.table_name = "test_mongo" # 指定存储的表名
  22.         item.title = title  # 给item属性赋值
  23.         yield item  # 返回item， item会自动批量入库
  26. if __name__ == "__main__":
  27.     TestMongo().start()

直接使用

连接

  1. from feapder.db.mongodb import MongoDB
  4. db = MongoDB(
  5.     ip="localhost", port=27017, db="feapder", user_name="feapder", user_pass="feapder123"
  6. )

若环境变量中配置了数据库连接方式或者setting中已配置，则可不传参 

  1. db = MongoDB()

或者可以根据url连接

  1. db = MongoDB.from_url("mongodb://username:password@ip:port/db")

方法

MongoDB封装了增删改查等方法，方便使用

查

  1. def find(self, table, limit=0) -> List[Dict]:
  2.     """
  3.     @summary:
  4.     无数据： 返回()
  5.     有数据： 若limit == 1 则返回 (data1, data2)
  6.             否则返回 ((data1, data2),)
  7.     ---------
  8.     @param table:
  9.     @param limit:
  10.     ---------
  11.     @result:
  12.     """

增

  1. def add(self, table, data, **kwargs):
  2.     """
  4.     Args:
  5.         table:
  6.         data:
  7.         kwargs:
  8.             auto_update: 覆盖更新，将替换唯一索引重复的数据，默认False
  9.             update_columns: 更新指定的列（如果数据的唯一索引存在，则更新指定字段，如 update_columns = ["name", "title"]
  10.             insert_ignore: 唯一索引冲突时是否忽略，默认为False
  11.             condition_fields: 用于条件查找的字段，默认以`_id`作为查找条件，默认：['_id']
  12.             exception_callfunc: 异常回调
  14.     Returns: 添加行数
  16.     """
  1. def add_batch(self, table: str, datas: List[Dict], **kwargs):
  2.     """
  3.     @summary: 批量添加数据
  4.     ---------
  5.     @param command: 字典
  6.     @param datas: 列表 [[..], [...]]
  7.     @param **kwargs:
  8.         auto_update: 覆盖更新，将替换唯一索引重复的数据，默认False
  9.         update_columns: 更新指定的列（如果数据的唯一索引存在，则更新指定字段，如 update_columns = ["name", "title"]
  10.         update_columns_value: 指定更新的字段对应的值
  11.         condition_fields: 用于条件查找的字段，默认以`_id`作为查找条件，默认：['_id']
  12.     ---------
  13.     @result: 添加行数
  14.     """

更新

  1. def update(self, coll_name, data: Dict, condition: Dict, upsert: bool = False):
  2.     """
  3.     更新
  4.     Args:
  5.         coll_name: 集合名
  6.         data: 单条数据 {"xxx":"xxx"}
  7.         condition: 更新条件 {"_id": "xxxx"}
  8.         upsert: 数据不存在则插入,默认为 False
  10.     Returns: True / False
  11.     """

删除

  1. def delete(self, table, condition: Dict):
  2.     """
  3.     删除
  4.     Args:
  5.         table:
  6.         condition: 查找条件
  7.     Returns: True / False
  8.     """