crawlstore

给 requests 等同步爬虫使用的轻量 MongoDB 落库组件。

安装

本地开发安装：

uv pip install -e .

或者从当前源码构建 wheel：

uv build

使用

from crawlstore import MongoSink

items = [
    {"url": "https://example.com/1", "title": "商品 1", "price": 19.9},
    {"url": "https://example.com/2", "title": "商品 2", "price": 29.9},
]

with MongoSink(
    uri="mongodb://127.0.0.1:27017",
    database="crawler",
) as sink:
    sink.ensure_indexes("products", unique=[("url",)])
    result = sink.save_many(
        items,
        collection="products",
        key_fields=("url",),
        mode="upsert",
    )
    print(result)

也可以使用环境变量：

$env:MONGO_URI = "mongodb://127.0.0.1:27017"
$env:MONGO_DATABASE = "crawler"

from crawlstore import MongoSink

sink = MongoSink.from_env()
sink.save(
    {"shop_id": "1001", "name": "示例店铺"},
    collection="shops",
    key_fields=("shop_id",),
)
sink.close()

读取数据

读取单条数据：

product = sink.find_one(
    collection="products",
    query={"url": "https://example.com/1"},
)

按条件批量读取：

products = sink.find_many(
    collection="products",
    query={"price": {"$gte": 10}},
    projection={"title": 1, "price": 1},
    sort=[("price", -1)],
    skip=0,
    limit=100,
)

for product in products:
    print(product)

统计数量：

total = sink.count(
    collection="products",
    query={"price": {"$gte": 10}},
)

upsert 会自动维护 _created_at 和 _updated_at，使用本地当前时间， 格式为 YYYY-MM-DD HH:MM，例如 2026-08-20 11:25。唯一键可以是组合字段：

sink.save_many(
    items,
    collection="products",
    key_fields=("platform", "product_id"),
)