Write a set of entities - given as dicts - to the index.
(
collection, entities, safe=False, role_id=None, mutable=True, clean=True
)
| 34 | |
| 35 | |
| 36 | def bulk_write( |
| 37 | collection, entities, safe=False, role_id=None, mutable=True, clean=True |
| 38 | ): |
| 39 | """Write a set of entities - given as dicts - to the index.""" |
| 40 | # This is called mainly by the /api/2/collections/X/_bulk API. |
| 41 | aggregator = get_aggregator(collection) |
| 42 | writer = aggregator.bulk() |
| 43 | for data in entities: |
| 44 | entity = model.get_proxy(data, cleaned=(not clean)) |
| 45 | entity = collection.ns.apply(entity) |
| 46 | if entity.id is None: |
| 47 | raise InvalidData("No ID for entity", errors=entity.to_dict()) |
| 48 | if safe: |
| 49 | entity = remove_checksums(entity) |
| 50 | entity.context = {"role_id": role_id, "mutable": mutable} |
| 51 | for field, func in (("created_at", min), ("updated_at", max)): |
| 52 | ts = func(ensure_list(data.get(field)), default=None) |
| 53 | dt = registry.date.to_datetime(ts) |
| 54 | if dt is not None: |
| 55 | entity.context[field] = dt.isoformat() |
| 56 | writer.put(entity, origin="bulk") |
| 57 | yield entity.id |
| 58 | writer.flush() |
no test coverage detected