diff --git a/gcloud/datastore/dataset.py b/gcloud/datastore/dataset.py
new file mode 100644
index 000000000000..16b2193c4041
--- /dev/null
+++ b/gcloud/datastore/dataset.py
@@ -0,0 +1,98 @@
+# Copyright 2014 Google Inc. All rights reserved.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+"""Convenience wrapper for invoking APIs/factories w/ a dataset ID."""
+
+from gcloud.datastore.api import delete
+from gcloud.datastore.api import get
+from gcloud.datastore.api import put
+from gcloud.datastore.batch import Batch
+from gcloud.datastore.key import Key
+from gcloud.datastore.query import Query
+from gcloud.datastore.transaction import Transaction
+
+
+class Dataset(object):
+    """Convenience wrapper for invoking APIs/factories w/ a dataset ID.
+
+    :type dataset_id: string
+    :param dataset_id: (required) dataset ID to pass to proxied API methods.
+
+    :type connection: :class:`gcloud.datastore.connection.Connection`, or None
+    :param connection: (optional) connection to pass to proxied API methods
+    """
+
+    def __init__(self, dataset_id, connection=None):
+        if dataset_id is None:
+            raise ValueError('dataset_id required')
+        self.dataset_id = dataset_id
+        self.connection = connection
+
+    def get(self, keys, missing=None, deferred=None):
+        """Proxy to :func:`gcloud.datastore.api.get`.
+
+        Passes our ``dataset_id``.
+        """
+        return get(keys, missing=missing, deferred=deferred,
+                   connection=self.connection, dataset_id=self.dataset_id)
+
+    def put(self, entities):
+        """Proxy to :func:`gcloud.datastore.api.put`.
+
+        Passes our ``dataset_id``.
+        """
+        return put(entities, connection=self.connection,
+                   dataset_id=self.dataset_id)
+
+    def delete(self, keys):
+        """Proxy to :func:`gcloud.datastore.api.delete`.
+
+        Passes our ``dataset_id``.
+        """
+        return delete(keys, connection=self.connection,
+                      dataset_id=self.dataset_id)
+
+    def key(self, *path_args, **kwargs):
+        """Proxy to :class:`gcloud.datastore.key.Key`.
+
+        Passes our ``dataset_id``.
+        """
+        if 'dataset_id' in kwargs:
+            raise TypeError('Cannot pass dataset_id')
+        kwargs['dataset_id'] = self.dataset_id
+        return Key(*path_args, **kwargs)
+
+    def batch(self):
+        """Proxy to :class:`gcloud.datastore.batch.Batch`.
+
+        Passes our ``dataset_id``.
+        """
+        return Batch(dataset_id=self.dataset_id, connection=self.connection)
+
+    def transaction(self):
+        """Proxy to :class:`gcloud.datastore.transaction.Transaction`.
+
+        Passes our ``dataset_id``.
+        """
+        return Transaction(dataset_id=self.dataset_id,
+                           connection=self.connection)
+
+    def query(self, **kwargs):
+        """Proxy to :class:`gcloud.datastore.query.Query`.
+
+        Passes our ``dataset_id``.
+        """
+        if 'dataset_id' in kwargs:
+            raise TypeError('Cannot pass dataset_id')
+        kwargs['dataset_id'] = self.dataset_id
+        return Query(**kwargs)
diff --git a/gcloud/datastore/test_dataset.py b/gcloud/datastore/test_dataset.py
new file mode 100644
index 000000000000..a51b7f33909e
--- /dev/null
+++ b/gcloud/datastore/test_dataset.py
@@ -0,0 +1,295 @@
+# Copyright 2014 Google Inc. All rights reserved.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import unittest2
+
+
+class TestDataset(unittest2.TestCase):
+
+    DATASET_ID = 'DATASET'
+
+    def _getTargetClass(self):
+        from gcloud.datastore.dataset import Dataset
+        return Dataset
+
+    def _makeOne(self, dataset_id=DATASET_ID, connection=None):
+        return self._getTargetClass()(dataset_id, connection)
+
+    def test_ctor_w_dataset_id_None(self):
+        self.assertRaises(ValueError, self._makeOne, None)
+
+    def test_ctor_w_dataset_id_no_connection(self):
+        dataset = self._makeOne()
+        self.assertEqual(dataset.dataset_id, self.DATASET_ID)
+
+    def test_ctor_w_dataset_id_w_connection(self):
+        conn = object()
+        dataset = self._makeOne(connection=conn)
+        self.assertEqual(dataset.dataset_id, self.DATASET_ID)
+        self.assertTrue(dataset.connection is conn)
+
+    def test_get_defaults(self):
+        from gcloud.datastore import dataset as MUT
+        from gcloud._testing import _Monkey
+
+        _called_with = []
+
+        def _get(*args, **kw):
+            _called_with.append((args, kw))
+
+        dataset = self._makeOne()
+        key = object()
+
+        with _Monkey(MUT, get=_get):
+            dataset.get([key])
+
+        self.assertEqual(_called_with[0][0], ([key],))
+        self.assertTrue(_called_with[0][1]['missing'] is None)
+        self.assertTrue(_called_with[0][1]['deferred'] is None)
+        self.assertTrue(_called_with[0][1]['connection'] is None)
+        self.assertEqual(_called_with[0][1]['dataset_id'], self.DATASET_ID)
+
+    def test_get_explicit(self):
+        from gcloud.datastore import dataset as MUT
+        from gcloud._testing import _Monkey
+
+        _called_with = []
+
+        def _get(*args, **kw):
+            _called_with.append((args, kw))
+
+        conn = object()
+        dataset = self._makeOne(connection=conn)
+        key, missing, deferred = object(), [], []
+
+        with _Monkey(MUT, get=_get):
+            dataset.get([key], missing, deferred)
+
+        self.assertEqual(_called_with[0][0], ([key],))
+        self.assertTrue(_called_with[0][1]['missing'] is missing)
+        self.assertTrue(_called_with[0][1]['deferred'] is deferred)
+        self.assertTrue(_called_with[0][1]['connection'] is conn)
+        self.assertEqual(_called_with[0][1]['dataset_id'], self.DATASET_ID)
+
+    def test_put_wo_connection(self):
+        from gcloud.datastore import dataset as MUT
+        from gcloud._testing import _Monkey
+
+        _called_with = []
+
+        def _put(*args, **kw):
+            _called_with.append((args, kw))
+
+        dataset = self._makeOne()
+        entity = object()
+
+        with _Monkey(MUT, put=_put):
+            dataset.put([entity])
+
+        self.assertEqual(_called_with[0][0], ([entity],))
+        self.assertTrue(_called_with[0][1]['connection'] is None)
+        self.assertEqual(_called_with[0][1]['dataset_id'], self.DATASET_ID)
+
+    def test_put_w_connection(self):
+        from gcloud.datastore import dataset as MUT
+        from gcloud._testing import _Monkey
+
+        _called_with = []
+
+        def _put(*args, **kw):
+            _called_with.append((args, kw))
+
+        entity, conn = object(), object()
+        dataset = self._makeOne(connection=conn)
+
+        with _Monkey(MUT, put=_put):
+            dataset.put([entity])
+
+        self.assertEqual(_called_with[0][0], ([entity],))
+        self.assertTrue(_called_with[0][1]['connection'] is conn)
+        self.assertEqual(_called_with[0][1]['dataset_id'], self.DATASET_ID)
+
+    def test_delete_wo_connection(self):
+        from gcloud.datastore import dataset as MUT
+        from gcloud._testing import _Monkey
+
+        _called_with = []
+
+        def _delete(*args, **kw):
+            _called_with.append((args, kw))
+
+        dataset = self._makeOne()
+        key = object()
+
+        with _Monkey(MUT, delete=_delete):
+            dataset.delete([key])
+
+        self.assertEqual(_called_with[0][0], ([key],))
+        self.assertTrue(_called_with[0][1]['connection'] is None)
+        self.assertEqual(_called_with[0][1]['dataset_id'], self.DATASET_ID)
+
+    def test_delete_w_connection(self):
+        from gcloud.datastore import dataset as MUT
+        from gcloud._testing import _Monkey
+
+        _called_with = []
+
+        def _delete(*args, **kw):
+            _called_with.append((args, kw))
+
+        key, conn = object(), object()
+        dataset = self._makeOne(connection=conn)
+        with _Monkey(MUT, delete=_delete):
+            dataset.delete([key])
+
+        self.assertEqual(_called_with[0][0], ([key],))
+        self.assertTrue(_called_with[0][1]['connection'] is conn)
+        self.assertEqual(_called_with[0][1]['dataset_id'], self.DATASET_ID)
+
+    def test_key_w_dataset_id(self):
+        KIND = 'KIND'
+        ID = 1234
+        dataset = self._makeOne()
+        self.assertRaises(TypeError,
+                          dataset.key, KIND, ID, dataset_id=self.DATASET_ID)
+
+    def test_key_wo_dataset_id(self):
+        from gcloud.datastore import dataset as MUT
+        from gcloud._testing import _Monkey
+        KIND = 'KIND'
+        ID = 1234
+        dataset = self._makeOne()
+
+        with _Monkey(MUT, Key=_Dummy):
+            key = dataset.key(KIND, ID)
+
+        self.assertTrue(isinstance(key, _Dummy))
+        self.assertEqual(key.args, (KIND, ID))
+        self.assertEqual(key.kwargs, {'dataset_id': self.DATASET_ID})
+
+    def test_batch_wo_connection(self):
+        from gcloud.datastore import dataset as MUT
+        from gcloud._testing import _Monkey
+        dataset = self._makeOne()
+
+        with _Monkey(MUT, Batch=_Dummy):
+            batch = dataset.batch()
+
+        self.assertTrue(isinstance(batch, _Dummy))
+        self.assertEqual(batch.args, ())
+        self.assertEqual(batch.kwargs,
+                         {'dataset_id': self.DATASET_ID, 'connection': None})
+
+    def test_batch_w_connection(self):
+        from gcloud.datastore import dataset as MUT
+        from gcloud._testing import _Monkey
+        conn = object()
+        dataset = self._makeOne(connection=conn)
+
+        with _Monkey(MUT, Batch=_Dummy):
+            batch = dataset.batch()
+
+        self.assertTrue(isinstance(batch, _Dummy))
+        self.assertEqual(batch.args, ())
+        self.assertEqual(batch.kwargs,
+                         {'dataset_id': self.DATASET_ID, 'connection': conn})
+
+    def test_transaction_wo_connection(self):
+        from gcloud.datastore import dataset as MUT
+        from gcloud._testing import _Monkey
+        dataset = self._makeOne()
+
+        with _Monkey(MUT, Transaction=_Dummy):
+            xact = dataset.transaction()
+
+        self.assertTrue(isinstance(xact, _Dummy))
+        self.assertEqual(xact.args, ())
+        self.assertEqual(xact.kwargs,
+                         {'dataset_id': self.DATASET_ID, 'connection': None})
+
+    def test_transaction_w_connection(self):
+        from gcloud.datastore import dataset as MUT
+        from gcloud._testing import _Monkey
+        conn = object()
+        dataset = self._makeOne(connection=conn)
+
+        with _Monkey(MUT, Transaction=_Dummy):
+            xact = dataset.transaction()
+
+        self.assertTrue(isinstance(xact, _Dummy))
+        self.assertEqual(xact.args, ())
+        self.assertEqual(xact.kwargs,
+                         {'dataset_id': self.DATASET_ID, 'connection': conn})
+
+    def test_query_w_dataset_id(self):
+        KIND = 'KIND'
+        dataset = self._makeOne()
+        self.assertRaises(TypeError,
+                          dataset.query, kind=KIND, dataset_id=self.DATASET_ID)
+
+    def test_query_w_defaults(self):
+        from gcloud.datastore import dataset as MUT
+        from gcloud._testing import _Monkey
+        dataset = self._makeOne()
+
+        with _Monkey(MUT, Query=_Dummy):
+            query = dataset.query()
+
+        self.assertTrue(isinstance(query, _Dummy))
+        self.assertEqual(query.args, ())
+        self.assertEqual(query.kwargs, {'dataset_id': self.DATASET_ID})
+
+    def test_query_explicit(self):
+        from gcloud.datastore import dataset as MUT
+        from gcloud._testing import _Monkey
+        KIND = 'KIND'
+        NAMESPACE = 'NAMESPACE'
+        ANCESTOR = object()
+        FILTERS = [('PROPERTY', '==', 'VALUE')]
+        PROJECTION = ['__key__']
+        ORDER = ['PROPERTY']
+        GROUP_BY = ['GROUPBY']
+        dataset = self._makeOne()
+
+        with _Monkey(MUT, Query=_Dummy):
+            query = dataset.query(
+                kind=KIND,
+                namespace=NAMESPACE,
+                ancestor=ANCESTOR,
+                filters=FILTERS,
+                projection=PROJECTION,
+                order=ORDER,
+                group_by=GROUP_BY,
+                )
+
+        self.assertTrue(isinstance(query, _Dummy))
+        kwargs = {
+            'dataset_id': self.DATASET_ID,
+            'kind': KIND,
+            'namespace': NAMESPACE,
+            'ancestor': ANCESTOR,
+            'filters': FILTERS,
+            'projection': PROJECTION,
+            'order': ORDER,
+            'group_by': GROUP_BY,
+        }
+        self.assertEqual(query.args, ())
+        self.assertEqual(query.kwargs, kwargs)
+
+
+class _Dummy(object):
+
+    def __init__(self, *args, **kwargs):
+        self.args = args
+        self.kwargs = kwargs