Basically I suggest you implement the Doctrine\Common\Collections\Collection interface, maybe extending ArrayCollection, and create a method save that will do what the doc told you to.
<?php
class MyDirtyCollection extends \Doctrine\Common\Collections\ArrayCollection {
public function __construct(AbstractService $abstractService)
{
$this->service = $abstractService;
}
public function save()
{
foreach ($this as $entity) {
$this->service->save($entity);
}
}
}
class MyCollection extends \Doctrine\Common\Collections\ArrayCollection {
public $bulkSize = 500;
protected $eventManager;
protected $entityManager;
public function __construct(EntityManager $entityManager, EventManager $eventManager)
{
$this->entityManager = $entityManager;
$this->eventManager = $eventManager;
}
public function getEventManager()
{
return $this->eventManager;
}
public function getEntityManager()
{
return $this->entityManager;
}
public function setBulkSize(int $bulkSize)
{
$this->bulkSize = $bulkSize;
}
public function save()
{
$transactionStarted = $this->getEntityManager()->getConnection()->beginTransaction();
try {
foreach ($this as $entity) {
$action = $entity->getId() ? self::UPDATE : self::CREATION;
$this->getEventManager()->trigger('save.pre', $entity, ['action' => $action]);
}
$i = 0;
foreach ($this as $entity) {
$i++;
$this->getEntityManager()->persist($entity);
if (($i % $this->bulkSize) === 0) {
$this->getEntityManager()->flush();
$this->getEntityManager()->clear();
}
}
$this->getEntityManager()->flush();
$this->getEntityManager()->clear();
foreach ($this as $entity) {
$action = $entity->getId() ? self::UPDATE : self::CREATION;
$this->getEventManager()->trigger('save.post', $entity, ['action' => $action]);
}
if ($transactionStarted) {
$this->getEntityManager()->getConnection()->commitTransaction();
}
} catch (Exception $e) {
$this->getEntityManager()->rollbackTransaction();
}
}
}
Something like that ;)
When you fetch your data you hydrate your collection, then you deal with your entity and finally call $collection->save();
EDIT : Add insert class and use case below :
The performance here will be low, but still better than commit by commit. Yet you should think about using Doctrine DBAL instead of the ORM if you are looking for hgih performance. Here I share with you my DBAL class for bulk Insert :
<?php
namespace JTH\Doctrine\DBAL;
use Doctrine\DBAL\Query\QueryBuilder;
use Exception;
use InvalidArgumentException;
use Traversable;
use UnderflowException;
class Insert extends QueryBuilder
{
const CALLBACK_FAILURE_SKIP = 0;
const CALLBACK_FAILURE_BREAK = 1;
protected $callbackFailureStrategy = self::CALLBACK_FAILURE_BREAK;
public static $defaultBulkSize = 500;
public $ignore = false;
public $onDuplicate = null;
public function values(array $values)
{
$this->resetQueryPart('values');
$this->addValues($values);
}
public function addValues(array $values)
{
$this->add('values', $values, true);
}
public function setCallbackFailureStrategy($strategy)
{
if ($strategy == static::CALLBACK_FAILURE_BREAK) {
$this->callbackFailureStrategy = static::CALLBACK_FAILURE_BREAK;
} elseif ($strategy == static::CALLBACK_FAILURE_SKIP) {
$this->callbackFailureStrategy = static::CALLBACK_FAILURE_SKIP;
} else {
$class = self::class;
throw new InvalidArgumentException(
"Invalid failure behaviour. See $class::CALLBACK_FAILURE_SKIP and $class::CALLBACK_FAILURE_BREAK"
);
}
}
public function getCallbackFailureStrategy()
{
return $this->callbackFailureStrategy;
}
public function execute()
{
return $this->getConnection()->executeUpdate(
$this->getSQLForInsert(),
$this->getParameters(),
$this->getParameterTypes()
);
}
/**
* Converts this instance into an INSERT string in SQL.
* @return string
* @throws \Exception
*/
private function getSQLForInsert()
{
$count = sizeof($this->getQueryPart('values'));
if ($count == 0) {
throw new UnderflowException("No values ready for INSERT");
}
$values = current($this->getQueryPart('values'));
$ignore = $this->ignore ? 'IGNORE' : '' ;
$sql = "INSERT $ignore INTO " . $this->getQueryPart('from')['table'] .
' (' . implode(', ', array_keys($values)) . ')' . ' VALUES ';
foreach ($this->getQueryPart('values') as $values) {
$sql .= '(' ;
foreach ($values as $value) {
if (is_array($value)) {
if ($value['raw']) {
$sql .= $value['value'] . ',';
} else {
$sql .= $this->expr()->literal($value['value'], $value['type']) . ',';
}
} else {
$sql .= $this->expr()->literal($value) . ',';
}
}
$sql = substr($sql, 0, -1);
$sql .= '),';
}
$sql = substr($sql, 0, -1);
if (!is_null($this->onDuplicate)) {
$sql .= ' ON DUPLICATE KEY UPDATE ' . $this->onDuplicate . ' ';
}
return $sql;
}
/**
* @param $loopable array | Traversable An array or object to loop over
* @param $callable Callable A callable that will be called before actually insert the row.
* two parameters will be passed :
* - the key of the current row
* - the row values (Array)
* An array of rows to insert must be returned
* @param $bulkSize int How many rows will be inserted at once
* @param bool $transactionnal
* @throws \Doctrine\DBAL\ConnectionException
* @throws \Exception
*/
public function bulk($loopable, callable $callable, $bulkSize = null, $transactionnal = true)
{
if (!is_array($loopable) and !($loopable instanceof Traversable)) {
throw new InvalidArgumentException("\$loppable must be either an array or a traversable object");
}
$bulkSize = $bulkSize ?? static::$defaultBulkSize;
$this->getConnection()->getConfiguration()->setSQLLogger(null); // Avoid MonoLog memory overload
if ($transactionnal) {
$this->getConnection()->beginTransaction();
}
$this->resetQueryPart('values');
foreach ($loopable as $key => $values) {
try {
$callbackedValues = $callable($key, $values);
if (sizeof($callbackedValues) > 0) {
foreach ($callbackedValues as $callbackedValuesRow) {
$this->addValues($callbackedValuesRow);
}
}
} catch (Exception $e) {
/*
* If a callback exception must break the transaction, then throw the exception to the call stack
* Else, skip the row insertion
*/
if ($this->callbackFailureStrategy == static::CALLBACK_FAILURE_BREAK) {
throw $e;
} else {
continue;
}
}
$count = count($this->getQueryPart('values'));
if ($count >= $bulkSize) {
$this->execute();
$this->resetQueryPart('values');
}
}
$count = count($this->getQueryPart('values'));
if ($count > 0) {
$this->execute();
}
$this->resetQueryPart('values');
if ($transactionnal) {
$this->getConnection()->commit();
}
}
/**
* @return boolean
*/
public function isIgnore()
{
return $this->ignore;
}
/**
* @param boolean $ignore
*/
public function setIgnore(bool $ignore)
{
$this->ignore = $ignore;
}
/**
* @return null|string
*/
public function getOnDuplicate() : string
{
return $this->onDuplicate;
}
/**
* @param null $onDuplicate
*/
public function setOnDuplicate($onDuplicate)
{
$this->onDuplicate = $onDuplicate;
$this->ignore = false;
}
}
Use case :
try {
$i = new Insert($this->getDoctrine()->getConnection('myDB'));
$i->insert('myTable');
$i->setOnDuplicate('col1 = VALUES(col1), updated_last = NOW()');
$i->setCallbackFailureStrategy(Insert::CALLBACK_FAILURE_BREAK);
$i->bulk($myArrayOfRows, function ($key, $row) {
// Some pre-insert processing
$rowset[] = $row;
return $rowset;
}, 500, true);
$this->addFlash('success', 'Yay !');
} catch (DBALException $e) {
$this->addFlash('error', 'Damn, error : ' . $e->getMessage());
}
