Method test_bulk_operations

benchmarks/test_sqlalchemy.py:310–347

Test bulk insert and update operations

(self)

            session.close()
    
    def test_bulk_operations(self):
        """Test bulk insert and update operations"""
        session = self.Session()
        try:
            # Bulk insert
            users = [
                User(username=f"bulk{i}", email=f"bulk{i}@example.com", age=20+i)
                for i in range(100)
            ]
            session.bulk_save_objects(users)
            session.commit()
            
            count = session.query(User).filter(User.username.like("bulk%")).count()
            assert count == 100
            
            # Bulk update
            session.query(User).filter(User.username.like("bulk%")).update(
                {User.is_active: False},
                synchronize_session=False
            )
            session.commit()
            
            inactive_count = session.query(User).filter(
                and_(User.username.like("bulk%"), User.is_active == False)
            ).count()
            assert inactive_count == 100
            
            # Bulk delete
            session.query(User).filter(User.username.like("bulk%")).delete(
                synchronize_session=False
            )
            session.commit()
            
            remaining = session.query(User).filter(User.username.like("bulk%")).count()
            assert remaining == 0
            
        finally:
            session.close()
    
    def test_datetime_operations(self):
        """Test datetime handling"""