Source code for couchbase.bucket
from couchbase.management import CollectionManager
from couchbase.management.admin import Admin
from couchbase_core.supportability import uncommitted
from couchbase_core.client import Client as CoreClient
from .collection import CBCollection, CollectionOptions
from .options import OptionBlock
from .result import *
from .collection import Scope
class BucketOptions(OptionBlock):
pass
class ViewOptions(OptionBlock):
pass
[docs]class Bucket(object):
_bucket = None # type: CoreClient
@overload
def __init__(self,
connection_string, # type: str
name=None,
admin=None, # type: Admin
*options # type: BucketOptions
):
# type: (...)->None
pass
[docs] def __init__(self,
connection_string, # type: str
name=None,
corebucket_class=CBCollection, # type: Type[CoreClient]
admin=None, # type: Admin
*options,
**kwargs
):
# type: (...)->None
"""
Connect to a bucket.
Typically invoked by :meth:`couchbase.cluster.Cluster.open_bucket`
:param str name: name of bucket.
:param str connection_string:
The connection string to use for connecting to the bucket.
This is a URI-like string allowing specifying multiple hosts
and a bucket name.
The format of the connection string is the *scheme*
(``couchbase`` for normal connections, ``couchbases`` for
SSL enabled connections); a list of one or more *hostnames*
delimited by commas; a *bucket* and a set of options.
like so::
couchbase://host1,host2,host3/bucketname?option1=value1&option2=value2
If using the SSL scheme (``couchbases``), ensure to specify
the ``certpath`` option to point to the location of the
certificate on the client's filesystem; otherwise connection
may fail with an error code indicating the server's
certificate could not be trusted.
See :ref:`connopts` for additional connection options.
:param string username: username to connect to bucket with
:param string password: the password of the bucket
:param boolean quiet: the flag controlling whether to raise an
exception when the client executes operations on
non-existent keys. If it is `False` it will raise
:exc:`.NotFoundError` exceptions. When
set to `True` the operations will return `None` silently.
:param boolean unlock_gil: If set (which is the default), the
bucket object will release the python GIL when possible,
allowing other (Python) threads to function in the
background. This should be set to true if you are using
threads in your application (and is the default), as
otherwise all threads will be blocked while couchbase
functions execute.
You may turn this off for some performance boost and you are
certain your application is not using threads
:param transcoder:
Set the transcoder object to use. This should conform to the
interface in the documentation (it need not actually be a
subclass). This can be either a class type to instantiate,
or an initialized instance.
:type transcoder: :class:`.Transcoder`
:param lockmode: The *lockmode* for threaded access.
See :ref:`multiple_threads` for more information.
:param tracer: An OpenTracing tracer into which
to propagate any tracing information. Requires
tracing to be enabled.
:raise: :exc:`.BucketNotFoundError` or :exc:`.AuthError` if
there is no such bucket to connect to, or if invalid
credentials were supplied.
:raise: :exc:`.CouchbaseNetworkError` if the socket wasn't
accessible (doesn't accept connections or doesn't respond
in
:raise: :exc:`.InvalidError` if the connection string
was malformed.
:return: instance of :class:`~couchbase.bucket.Bucket`
"""
self._name = name
self._connstr=connection_string
self._bucket_args=forward_args(kwargs, *options)
self._bucket_args['bucket']=name
self._corebucket_class=corebucket_class
self._bucket = CoreClient(connection_string, **self._bucket_args)
self._admin = admin
@property
def name(self):
# type: (...)->str
return self._name
[docs] @uncommitted
def scope(self,
scope_name # type: str
):
"""
Open the named scope.
:param scope_name:
:return: the named scope
:rtype: Scope
"""
# type: (...)->Scope
return Scope(self, scope_name)
[docs] def default_collection(self,
options=None # type: CollectionOptions
):
# type: (...)->CBCollection
"""
Open the default collection.
:param CollectionOptions options: any options to pass to the Collection constructor
:return: the default :class:`Collection` object.
"""
return Scope(self).default_collection()
[docs] @uncommitted
def collection(self,
collection_name, # type: str
options=None # type: CollectionOptions
):
# type: (...)->CBCollection
"""
Open a collection in the default scope.
:param collection_name: collection name
:param CollectionOptions options: any options to pass to the Collection constructor
:return: the default :class:`.Collection` object.
"""
return Scope(self).collection(collection_name)
def collections(self):
return CollectionManager(self._admin, self._name)
def view_query(self,
design_doc, # type: str
view_name, # type: str
*view_options # type: ViewOptions
):
# type: (...)->IViewResult
"""
Run a View Query
:param str design_doc: design document
:param str view_name: view name
:param view_options:
:return: IViewResult containing the view results
"""
cb = self._bucket # type: CoreClient
res = cb.query(design_doc, view_name, **forward_args(None, *view_options))
return IViewResult(res)
def views(self):
# type: (...)->IViewManager
pass
def ping(self,
options=None # type: PingOptions
):
# type: (...)->IPingResult
pass