Method insert_embeddings

deepface/modules/database/mongo.py:174–233

Insert embeddings into MongoDB. Args: embeddings (List[Dict[str, Any]]): List of embedding records to insert. batch_size (int): Number of records to insert in each batch. Returns: int: Number of embeddings successfully inserted.

(self, embeddings: List[Dict[str, Any]], batch_size: int = 100)

        return bytes(doc["index_data"])

    def insert_embeddings(self, embeddings: List[Dict[str, Any]], batch_size: int = 100) -> int:
        """
        Insert embeddings into MongoDB.
        Args:
            embeddings (List[Dict[str, Any]]): List of embedding records to insert.
            batch_size (int): Number of records to insert in each batch.
        Returns:
            int: Number of embeddings successfully inserted.
        """
        if not embeddings:
            raise ValueError("No embeddings to insert.")

        docs: List[Dict[str, Any]] = []

        for e in embeddings:
            face = e["face"]
            face_shape = list(face.shape)

            binary_face_data = self.Binary(face.astype(np.float32).tobytes())

            embedding_bytes = struct.pack(f'{len(e["embedding"])}d', *e["embedding"])

            face_hash = hashlib.sha256(json.dumps(face.tolist()).encode()).hexdigest()
            embedding_hash = hashlib.sha256(embedding_bytes).hexdigest()

            int_id = self.counters.find_one_and_update(
                {"_id": "embedding_id"}, {"$inc": {"seq": 1}}, upsert=True, return_document=True
            )["seq"]

            docs.append(
                {
                    "sequence": int_id,
                    "img_name": e["img_name"],
                    "face": binary_face_data,
                    "face_shape": face_shape,
                    "model_name": e["model_name"],
                    "detector_backend": e["detector_backend"],
                    "aligned": e["aligned"],
                    "l2_normalized": e["l2_normalized"],
                    "embedding": e["embedding"],
                    "face_hash": face_hash,
                    "embedding_hash": embedding_hash,
                    "created_at": datetime.now(timezone.utc),
                }
            )

        inserted = 0
        try:
            for i in range(0, len(docs), batch_size):
                result = self.embeddings.insert_many(docs[i : i + batch_size], ordered=False)
                inserted += len(result.inserted_ids)
        except (self.DuplicateKeyError, self.BulkWriteError) as e:
            if len(docs) == 1:
                logger.warn("Duplicate detected for extracted face and embedding.")
                return inserted
            raise DuplicateEntryError(
                f"Duplicate detected for extracted face and embedding in {i}-th batch"
            ) from e