1"""Native storage for OmegaML using mongodb as the storage layer
2
3An OmegaStore instance is a MongoDB database. It has at least the
4metadata collection which lists all objects stored in it. A metadata
5document refers to the following types of objects (metadata.kind):
6
7* pandas.dfrows - a Pandas DataFrame stored as a collection of rows
8* sklearn.joblib - a scikit learn estimator/pipline dumped using joblib.dump()
9* python.data - an arbitrary python dict, tuple, list stored as a document
10
11Note that storing Pandas and scikit learn objects requires the availability
12of the respective packages. If either can not be imported, the OmegaStore
13degrades to a python.data store only. It will still .list() and get() any
14object, however reverts to pure python objects. In this case it is up
15to the client to convert the data into an appropriate format for processing.
16
17Pandas and scikit-learn objects can only be stored if these packages are
18availables. put() raises a TypeError if you pass such objects and these
19modules cannot be loaded.
20
21All data are stored within the same mongodb, in per-object collections
22as follows:
23
24 * .metadata
25 all metadata. each object is one document,
26 See **omegaml.documents.Metadata** for details
27 * .<bucket>.files
28 this is the GridFS instance used to store
29 blobs (models, numpy, hdf). The actual file name
30 will be <prefix>/<name>.<ext>, where ext is
31 optionally generated by put() / get().
32 * .<bucket>.<prefix>.<name>.data
33 every other dataset is stored in a separate
34 collection (dataframes, dicts, lists, tuples).
35 Any forward slash in prefix is ignored (e.g. 'data/'
36 becomes 'data')
37
38 DataFrames by default are stored in their own collection, every
39 row becomes a document. To store dataframes as a binary file,
40 use `put(...., as_hdf=True).` `.get()` will always return a dataframe.
41
42 Python dicts, lists, tuples are stored as a single document with
43 a `.data` attribute holding the JSON-converted representation. `.get()`
44 will always return the corresponding python object of .data.
45
46 Models are joblib.dump()'ed and ziped prior to transferring into
47 GridFs. .get() will always unzip and joblib.load() before returning
48 the model. Note this requires that the process using .get() supports
49 joblib as well as all python classes referred to. If joblib is not
50 supported, .get() returns a file-like object.
51
52 The .metadata entry specifies the format used to store each
53 object as well as it's location:
54
55 * metadata.kind
56 the type of object
57 * metadata.name
58 the name of the object, as given on put()
59 * metadata.gridfile
60 the gridfs object (if any, null otherwise)
61 * metadata.collection
62 the name of the collection
63 * metadata.attributes
64 arbitrary custom attributes set in
65 put(attributes=obj). This is used e.g. by
66 OmegaRuntime's fit() method to record the
67 data used in the model's training.
68
69 **.put()** and **.get()** use helper methods specific to the type in
70 object's type and metadata.kind, respectively. In the future
71 a plugin system will enable extension to other types.
72"""
73
74import logging
75import os
76import shutil
77import warnings
78import weakref
79from uuid import uuid4
80
81import bson
82import gridfs
83from mongoengine.connection import connect, disconnect, get_db
84from mongoengine.errors import DoesNotExist
85from mongoengine.queryset.visitor import Q
86
87from omegaml.documents import MDREGISTRY, make_Metadata
88from omegaml.mongoshim import sanitize_mongo_kwargs, waitForConnection
89from omegaml.util import PickableCollection, ensure_index, extend_instance, load_class, signature, urlparse
90from omegaml.util import settings as omega_settings
91
92logger = logging.getLogger(__name__)
93
94
[docs]
95class OmegaStore:
96 """The storage backend for models and data
97
98 .. versionchanged:: 0.18.0
99 refactored all methods handling Python and Pandas datatypes to omegaml.backends.coreobjects.CoreObjectsBackend
100 """
101
102 def __init__(self, mongo_url=None, bucket=None, prefix=None, kind=None, defaults=None, dbalias=None):
103 """
104 :param mongo_url: the mongourl to use for the gridfs
105 :param bucket: the mongo collection to use for gridfs
106 :param prefix: the path prefix for files. defaults to blank
107 :param kind: the kind or list of kinds to limit this store to
108 """
109 self.defaults = defaults or omega_settings()
110 self.mongo_url = mongo_url or self.defaults.OMEGA_MONGO_URL
111 self.bucket = bucket or self.defaults.OMEGA_MONGO_COLLECTION
112 self._fs_collection = self._ensure_fs_collection()
113 self._fs = None
114 self._tmppath = os.path.join(self.defaults.OMEGA_TMP, uuid4().hex)
115 self.prefix = prefix or ''
116 self.force_kind = kind
117 self._Metadata_cls = None
118 # don't initialize db here to avoid using the default settings
119 # otherwise Metadata will already have a connection and not use
120 # the one provided in override_settings
121 self._db = None
122 self._dbalias = dbalias
123 # add backends and mixins
124 self._apply_mixins()
125 # register backends
126 self.register_backends()
127 # register finalizer for cleanup
128 weakref.finalize(self, self._cleanup, repr(self), tmppath=str(self._tmppath))
129
130 def __repr__(self):
131 return f'OmegaStore(bucket={self.bucket}, prefix={self.prefix})'
132
133 def __equal__(self, other):
134 """test for equality of OmegaStore instances
135
136 Args:
137 other: OmegaStore instance
138
139 Returns:
140 True if other is the same database, same bucket, same prefix
141 """
142 return self.mongo_url == other.mongo_url and self.bucket == other.bucket and self.prefix == other.prefix
143
144 @property
145 def tmppath(self):
146 """
147 return an instance-specific temporary path
148 """
149 os.makedirs(self._tmppath, exist_ok=True)
150 return self._tmppath
151
152 @property
153 def mongodb(self):
154 """
155 Returns a mongo database object
156 """
157 if self._db is not None:
158 return self._db
159 # parse salient parts of mongourl, e.g.
160 # mongodb://user:password@host/dbname
161 self.parsed_url = urlparse.urlparse(self.mongo_url)
162 self.database_name = self.parsed_url.path[1:]
163 host = self.parsed_url.netloc
164 scheme = self.parsed_url.scheme
165 username, password = None, None
166 if '@' in host:
167 creds, host = host.split('@', 1)
168 if ':' in creds:
169 username, password = creds.split(':')
170 # connect via mongoengine
171 #
172 # note this uses a MongoClient in the background, with pooled
173 # connections. there are multiprocessing issues with pymongo:
174 # http://api.mongodb.org/python/3.2/faq.html#using-pymongo-with-multiprocessing
175 # connect=False is due to https://jira.mongodb.org/browse/PYTHON-961
176 # this defers connecting until the first access
177 # serverSelectionTimeoutMS=2500 is to fail fast, the default is 30000
178 #
179 # use an instance specific alias, note that access to Metadata and
180 # QueryCache must pass the very same alias
181 self._dbalias = alias = self._dbalias or f'omega-{uuid4().hex}'
182 # local import of _connections ensure we get the actual object, not an earlier version
183 from mongoengine.connection import _connections
184
185 if alias not in _connections:
186 # always disconnect before registering a new connection because
187 # mongoengine.connect() forgets all connection settings upon disconnect
188 disconnect(alias)
189 connection = connect(
190 alias=alias,
191 db=self.database_name,
192 host=f'{scheme}://{host}',
193 username=username,
194 password=password,
195 connect=False,
196 authentication_source='admin',
197 **sanitize_mongo_kwargs(self.defaults.OMEGA_MONGO_SSL_KWARGS),
198 )
199 # since PyMongo 4, connect() no longer waits for connection
200 waitForConnection(connection)
201 self._db = get_db(alias)
202 return self._db
203
204 @property
205 def _Metadata(self):
206 if self._Metadata_cls is None:
207 # hack to localize metadata
208 db = self.mongodb
209 self._Metadata_cls = make_Metadata(db_alias=self._dbalias, collection=self._fs_collection)
210 return self._Metadata_cls
211
212 @property
213 def fs(self):
214 """
215 Retrieve a gridfs instance using url and collection provided
216
217 :return: a gridfs instance
218 """
219 if self._fs is not None:
220 return self._fs
221 self._fs = gridfs.GridFS(self.mongodb, collection=self._fs_collection)
222 self._ensure_fs_index(self._fs)
223 return self._fs
224
241
266
267 def _make_metadata(self, name=None, bucket=None, prefix=None, **kwargs):
268 """
269 create or update a metadata object
270
271 this retrieves a Metadata object if it exists given the kwargs. Only
272 the name, prefix and bucket arguments are considered
273
274 for existing Metadata objects, the attributes kw is treated as follows:
275
276 * attributes=None, the existing attributes are left as is
277 * attributes={}, the attributes value on an existing metadata object
278 is reset to the empty dict
279 * attributes={ some : value }, the existing attributes are updated
280
281 For new metadata objects, attributes defaults to {} if not specified,
282 else is set as provided.
283
284 :param name: the object name
285 :param bucket: the bucket, optional, defaults to self.bucket
286 :param prefix: the prefix, optional, defaults to self.prefix
287 """
288 bucket = bucket or self.bucket
289 prefix = prefix or self.prefix
290 meta = self.metadata(name=name, prefix=prefix, bucket=bucket)
291 if meta:
292 dict_fields = 'attributes', 'kind_meta'
293 for k, v in kwargs.items():
294 if k in dict_fields and v is not None and len(v) > 0:
295 previous = getattr(meta, k, {})
296 previous.update(v)
297 setattr(meta, k, previous)
298 elif k in dict_fields and v is not None and len(v) == 0:
299 setattr(meta, k, {})
300 elif k in dict_fields and v is None:
301 # ignore non specified attributes
302 continue
303 else:
304 # by default set whatever attribute is provided
305 setattr(meta, k, v)
306 else:
307 meta = self._Metadata(name=name, bucket=bucket, prefix=prefix, **kwargs)
308 return meta
309
310 def _drop_metadata(self, name=None, **kwargs):
311 # internal method to delete meta data of an object
312 meta = self.metadata(name, **kwargs)
313 if meta is not None:
314 meta.delete()
315
[docs]
316 def collection(self, name=None, bucket=None, prefix=None):
317 """
318 Returns a mongo db collection as a datastore
319
320 If there is an existing object of name, will return the .collection
321 of the object. Otherwise returns the collection according to naming
322 convention {bucket}.{prefix}.{name}.datastore
323
324 :param name: the collection to use. if none defaults to the
325 collection name given on instantiation. the actual collection name
326 used is always prefix + name + '.data'
327 """
328 # see if we have a known object and a collection for that, if not define it
329 meta = self.metadata(name, bucket=bucket, prefix=prefix)
330 collection = meta.collection if meta else None
331 if not collection:
332 collection = self.object_store_key(name, '.datastore')
333 collection = collection.replace('..', '.')
334 # return the collection
335 try:
336 datastore = getattr(self.mongodb, collection)
337 except Exception as e:
338 raise e
339 return PickableCollection(datastore)
340
341 def _apply_mixins(self):
342 """
343 apply mixins in defaults.OMEGA_STORE_MIXINS
344 """
345 for mixin in self.defaults.OMEGA_STORE_MIXINS:
346 conditional = self._mixins_conditional
347 extend_instance(self, mixin, conditional=conditional)
348
349 def _mixins_conditional(self, cls, obj):
350 return cls.supports(obj) if hasattr(cls, 'supports') else True
351
[docs]
352 def register_backends(self):
353 """
354 register backends in defaults.OMEGA_STORE_BACKENDS
355 """
356 # enable list modification within loop
357 # -- avoid RuntimeError: dictionary changed size during iteration
358 backends = list(self.defaults.OMEGA_STORE_BACKENDS.items())
359 for kind, backend in backends:
360 self.register_backend(kind, backend)
361
[docs]
362 def register_backend(self, kind, backend, index=-1):
363 """
364 register a backend class
365
366 :param kind: (str) the backend kind
367 :param backend: (class) the backend class
368 :param index: (int) the insert position, defaults to -1, which means
369 to append
370
371 .. versionchanged:: 0.18.0
372 added index to have more control over backend evaluation by .get_backend_byobj()
373
374 .. versionchanged:: 0.18.0
375 backends can specify cls.KIND_EXT to register additional kinds
376 """
377 backend_cls = load_class(backend)
378 backend_kinds = [backend_cls.KIND] + list(getattr(backend_cls, 'KIND_EXT', []))
379 for kind in backend_kinds:
380 self.defaults.OMEGA_STORE_BACKENDS[kind] = backend_cls
381 if kind not in MDREGISTRY.KINDS:
382 pos = len(MDREGISTRY.KINDS) + index if index < 0 else index
383 MDREGISTRY.KINDS.insert(pos, kind)
384 return self
385
[docs]
386 def register_mixin(self, mixincls):
387 """
388 register a mixin class
389
390 :param mixincls: (class) the mixin class
391 """
392 extend_instance(self, mixincls)
393 return self
394
[docs]
395 def put(self, obj, name, attributes=None, kind=None, replace=False, model_store=None, data_store=None, **kwargs):
396 """
397 Stores an object, store estimators, pipelines, numpy arrays or
398 pandas dataframes
399 """
400 if replace:
401 self.drop(name, force=True)
402 backend = self.get_backend_byobj(
403 obj, name, attributes=attributes, kind=kind, model_store=model_store, data_store=data_store, **kwargs
404 )
405 if backend:
406 return backend.put(obj, name, attributes=attributes, **kwargs)
407 raise TypeError('type %s not supported' % type(obj))
408
[docs]
409 def drop(self, name, force=False, version=-1, report=False, **kwargs):
410 """
411 Drop the object
412
413 :param name: The name of the object. If the name is a pattern it will
414 be expanded using .list(), and call .drop() on every obj found.
415 :param force: If True ignores DoesNotExist exception, defaults to False
416 meaning this raises a DoesNotExist exception if the name does not
417 exist
418 :param report: if True returns a dict name=>status, where status is True
419 if deleted, False if not deleted
420 :return: True if object was deleted, False if not.
421 If force is True and the object does not exist it will still return True
422 :raises: DoesNotExist if the object does not exist and ```force=False```
423 """
424 is_pattern = '*' in name
425 objs = [name] if not is_pattern else self.list(name)
426 results = []
427 for name in objs:
428 try:
429 backend = self.get_backend(name)
430 drop = backend.drop if hasattr(backend, 'drop') else self._drop
431 result = drop(name, force=force, version=version, **kwargs)
432 except Exception:
433 result = False
434 if not force and not is_pattern:
435 raise
436 if force:
437 result = self._drop(name, force=force, version=version)
438 results.append((name, result))
439 if not objs:
440 result = self._drop(name, force=force, version=version)
441 results.append((name, result))
442 return dict(results) if report else len(results) > 0
443
444 def _drop(self, name, force=False, version=-1, keep_data=False, **kwargs):
445 meta = self.metadata(name, version=version)
446 if meta is None and not force:
447 raise DoesNotExist(name)
448 collection = self.collection(name)
449 if collection and not keep_data:
450 self.mongodb.drop_collection(collection.name)
451 if meta:
452 if meta.collection and not keep_data:
453 self.mongodb.drop_collection(meta.collection)
454 if meta and meta.gridfile is not None and not keep_data:
455 meta.gridfile.delete()
456 self._drop_metadata(name)
457 return True
458 return False
459
[docs]
460 def get_backend_bykind(self, kind, model_store=None, data_store=None, **kwargs):
461 """
462 return the backend by a given object kind
463
464 :param kind: The object kind
465 :param model_store: the OmegaStore instance used to store models
466 :param data_store: the OmegaStore instance used to store data
467 :param kwargs: the kwargs passed to the backend initialization
468 :return: the backend
469 """
470 try:
471 backend_cls = load_class(self.defaults.OMEGA_STORE_BACKENDS[kind])
472 except KeyError:
473 raise ValueError('backend {kind} does not exist'.format(**locals()))
474 model_store = model_store or self
475 data_store = data_store or self
476 backend = backend_cls(model_store=model_store, data_store=data_store, **kwargs)
477 return backend
478
[docs]
479 def get_backend(self, name, model_store=None, data_store=None, kind=None, **kwargs):
480 """
481 return the backend by a given object name
482
483 :param kind: The object kind
484 :param model_store: the OmegaStore instance used to store models
485 :param data_store: the OmegaStore instance used to store data
486 :param kwargs: the kwargs passed to the backend initialization
487 :return: the backend
488 """
489 meta = self.metadata(name)
490 kind = kind or meta.kind if meta is not None else None
491 if kind in self.defaults.OMEGA_STORE_BACKENDS:
492 return self.get_backend_bykind(kind, model_store=model_store, data_store=data_store, **kwargs)
493 return None
494
[docs]
495 def help(self, name_or_obj=None, kind=None, raw=False, display=None, renderer=None):
496 """get help for an object by looking up its backend and calling help() on it
497
498 Retrieves the object's metadata and looks up its corresponding backend. If the
499 metadata.attributes['docs'] is a string it will display this as the help() contents.
500 If the string starts with 'http://' or 'https://' it will open the web page.
501
502 Args:
503 name_or_obj (str|obj): the name or actual object to get help for
504 kind (str): optional, if specified forces retrieval of backend for the given kind
505 raw (bool): optional, if True forces help to be the backend type of the object.
506 If False returns the attributes[docs] on the object's metadata, if available.
507 Defaults to False
508 display (fn): optional, callable for interactive display, defaults to help in
509 if sys.flags.interactive is True, else uses pydoc.render_doc with plaintext
510 renderer (fn): optional, the renderer= argument for pydoc.render_doc to use if
511 sys.flags.interactive is False and display is not provided
512
513 Returns:
514 * help(obj) if python is in interactive mode
515 * text(str) if python is in not interactive mode
516 """
517 import pydoc
518 import sys
519
520 interactive = bool(display) if display is not None else sys.flags.interactive
521 display = display or help
522 renderer = renderer or pydoc.plaintext
523 obj = self._resolve_help_backend(name_or_obj=name_or_obj, kind=kind, raw=raw)
524 if any(str(obj.__doc__).startswith(v) for v in ('https://', 'http://')):
525 obj = obj.__doc__
526 if interactive and display is help:
527 import webbrowser
528
529 display = webbrowser.open
530 return display(obj) if interactive else pydoc.render_doc(obj, renderer=renderer)
531
532 def _resolve_help_backend(self, name_or_obj=None, kind=None, raw=False):
533 # helper so we can test help
534 meta = self.metadata(name_or_obj) if name_or_obj else None
535 if kind:
536 backend = self.get_backend_bykind(kind)
537 else:
538 backend = self.get_backend(name_or_obj) or self.get_backend_byobj(name_or_obj)
539 if backend is None:
540 backend = self.get_backend_bykind('core.object', model_store=self, data_store=self)
541 if not raw and meta is not None and 'docs' in meta.attributes:
542
543 def UserDocumentation():
544 pass
545
546 basedoc = backend.__doc__ or ''
547 UserDocumentation.__doc__ = basedoc + meta.attributes['docs']
548 backend = UserDocumentation
549 return backend
550
[docs]
551 def get_backend_byobj(
552 self, obj, name=None, kind=None, attributes=None, model_store=None, data_store=None, **kwargs
553 ):
554 """
555 return the matching backend for the given obj
556
557 Returns:
558 the first backend that supports the given parameters or None
559
560 .. versionchanged:: 0.18.0
561 backends are tested in order of MDREGISTRY.KINDS, see .register_backend()
562 """
563 model_store = model_store or self
564 data_store = data_store or self
565 meta = self.metadata(name) if name else None
566 kind = kind or (meta.kind if meta is not None else None)
567 backend = None
568 if kind:
569 if kind in self.defaults.OMEGA_STORE_BACKENDS:
570 backend = self.get_backend_bykind(kind, data_store=data_store, model_store=model_store)
571 if not backend.supports(
572 obj,
573 name,
574 attributes=attributes,
575 data_store=data_store,
576 model_store=model_store,
577 meta=meta,
578 kind=kind,
579 **kwargs,
580 ):
581 objtype = str(type(obj))
582 warnings.warn('Backend {kind} does not support {objtype}'.format(**locals()))
583 else:
584 # revert to core backend
585 backend = self.get_backend_bykind('core.object', model_store=model_store, data_store=data_store)
586 else:
587 # sort by order in MDREGISTRY.KINDS, only using kinds that actually have a registered backend
588 sorted_backends = (k for k in MDREGISTRY.KINDS if k in self.defaults.OMEGA_STORE_BACKENDS)
589 for backend_kind in sorted_backends:
590 backend = self.get_backend_bykind(backend_kind, data_store=data_store, model_store=model_store)
591 if backend.supports(
592 obj, name, attributes=attributes, data_store=data_store, model_store=model_store, **kwargs
593 ):
594 break
595 else:
596 backend = None
597 return backend
598
[docs]
599 def getl(self, *args, **kwargs):
600 """return a lazy MDataFrame for a given object
601
602 Same as .get, but returns a MDataFrame
603
604 """
605 return self.get(*args, lazy=True, **kwargs)
606
[docs]
607 def get(self, name, version=-1, force_python=False, kind=None, model_store=None, data_store=None, **kwargs):
608 """
609 Retrieve an object
610
611 :param name: The name of the object
612 :param version: Version of the stored object (not supported)
613 :param force_python: Return as a python object
614 :param kwargs: kwargs depending on object kind
615 :return: an object, estimator, pipelines, data array or pandas dataframe
616 previously stored with put()
617 """
618 meta = self.metadata(name, version=version)
619 if meta is None:
620 return None
621 if not force_python:
622 backend = (
623 self.get_backend(name, model_store=model_store, data_store=data_store, **kwargs)
624 if not kind
625 else self.get_backend_bykind(kind, model_store=model_store, data_store=data_store, **kwargs)
626 )
627 if backend is not None:
628 # FIXME: some backends need to get model_store, data_store, but fails tests
629 return backend.get(name, **kwargs) # model_store=model_store, data_store=data_store, **kwargs)
630 # catch-call to CoreObjectsBackend or force python
631 # -- keeping the same behavior until version 0.17, handling all other KINDs
632 core_backend = self.get_backend_bykind('core.object', model_store=model_store, data_store=data_store, **kwargs)
633 if force_python:
634 return core_backend.get_object_as_python(meta, version=version)
635 return core_backend.get(name, version=version, force_python=force_python, **kwargs)
636
637 def __iter__(self):
638 for f in self.list(include_temp=True):
639 yield f
640
641 def __contains__(self, item):
642 """
643
644 Args:
645 item (str|Metadata): the item to be found using OmegaStore.list(pattern=item, include_hidden=True).
646 If the item is a Metadata instance, will use the item.name and item.kind
647
648 Returns:
649 bool: True if item can be found in this store
650 """
651 if isinstance(item, self._Metadata):
652 pattern, kind = item.name, item.kind
653 else:
654 pattern, kind = item, None
655 return item in self.list(pattern, kind=kind, hidden=True, include_temp=True)
656
657 @property
658 def buckets(self):
659 return [
660 'default' if b == self.defaults.OMEGA_MONGO_COLLECTION else b
661 for b in self._Metadata.objects.distinct('bucket')
662 ]
663
[docs]
664 def list(
665 self,
666 pattern=None,
667 regexp=None,
668 kind=None,
669 raw=False,
670 hidden=None,
671 include_temp=False,
672 bucket=None,
673 prefix=None,
674 filter=None,
675 ):
676 """
677 List all files in store
678
679 specify pattern as a unix pattern (e.g. :code:`models/*`,
680 or specify regexp)
681
682 :param pattern: the unix file pattern or None for all
683 :param regexp: the regexp. takes precedence over pattern
684 :param raw: if True return the meta data objects
685 :param filter: specify additional filter criteria, optional
686 :return: List of files in store
687
688 """
689 regex = lambda pattern: bson.regex.Regex(f'{pattern}')
690 db = self.mongodb
691 searchkeys = dict(bucket=bucket or self.bucket, prefix=prefix or self.prefix)
692 q_excludes = Q()
693 if regexp:
694 searchkeys['name'] = regex(regexp)
695 elif pattern:
696 re_pattern = pattern.replace('*', '.*').replace('/', r'\/')
697 searchkeys['name'] = regex(f'^{re_pattern}$')
698 if not include_temp:
699 q_excludes &= Q(name__not__startswith='_')
700 q_excludes &= Q(name__not=regex(r'(.{1,*}\/?_.*)'))
701 if not hidden:
702 q_excludes &= Q(name__not__startswith='.')
703 if kind or self.force_kind:
704 kind = kind or self.force_kind
705 if isinstance(kind, (tuple, list)):
706 searchkeys.update(kind__in=kind)
707 else:
708 searchkeys.update(kind=kind)
709 if filter:
710 searchkeys.update(filter)
711 q_search = Q(**searchkeys) & q_excludes
712 files = self._Metadata.objects.no_cache()(q_search)
713 return [f if raw else str(f.name).replace('.omm', '') for f in files]
714
[docs]
715 def exists(self, name, hidden=False):
716 """check if object exists
717
718 Args:
719 name (str): name of object
720 hidden (bool): if True, include hidden files, defaults to
721 False, unless name starts with '.'
722
723 Returns:
724 bool, True if object exists
725
726 .. versionchanged:: 0.16.4
727 hidden defaults to True if name starts with '.'
728 """
729 hidden = True if name.startswith('.') else hidden
730 return name in self.list(name, hidden=hidden)
731
[docs]
732 def object_store_key(self, name, ext, hashed=None):
733 """
734 Returns the store key
735
736 Unless you write a mixin or a backend you should not use this method
737
738 :param name: The name of object
739 :param ext: The extension of the filename
740 :param hashed: hash the key to support arbitrary name length, defaults
741 to defaults.OMEGA_STORE_HASHEDNAMES, True by default since 0.13.7
742
743 :return: A filename with relative bucket, prefix and name
744 """
745 # byte string
746 _u8 = lambda t: t.encode('UTF-8', 'replace') if isinstance(t, str) else t
747 key = self._get_obj_store_key(name, ext)
748 hashed = hashed if hashed is not None else self.defaults.OMEGA_STORE_HASHEDNAMES
749 if hashed:
750 from hashlib import sha1
751
752 # SEC: CWE-916
753 # - status: wontfix
754 # - reason: hashcode is used purely for name resolution, not a security function
755 hasher = sha1()
756 hasher.update(_u8(key))
757 key = hasher.hexdigest()
758 return key
759
760 def _get_obj_store_key(self, name, ext, prefix=None, bucket=None):
761 # backwards compatilibity implementation of object_store_key()
762 name = '%s.%s' % (name, ext) if not name.endswith(ext) else name
763 filename = f'{bucket or self.bucket}.{prefix or self.prefix}.{name}'.replace('/', '_').replace('..', '.')
764 return filename
765
766 def _ensure_fs_collection(self):
767 # ensure backwards-compatible gridfs access
768 if self.defaults.OMEGA_BUCKET_FS_LEGACY:
769 # prior to 0.13.2 a single gridfs instance was used, always equal to the default collection
770 return self.defaults.OMEGA_MONGO_COLLECTION
771 if self.bucket == self.defaults.OMEGA_MONGO_COLLECTION:
772 # from 0.13.2 onwards, only the default bucket is equal to the default collection
773 # backwards compatibility for existing installations
774 return self.bucket
775 # since 0.13.2, all buckets other than the default use a qualified collection name to
776 # effectively separate files in different buckets, enabling finer-grade access control
777 # and avoiding name collisions from different buckets
778 return f'{self.defaults.OMEGA_MONGO_COLLECTION}_{self.bucket}'
779
780 def _ensure_fs_index(self, fs):
781 # make sure we have proper chunks and file indicies. this should be created on first write, but sometimes is not
782 # see https://docs.mongodb.com/manual/core/gridfs/#gridfs-indexes
783 # pymongo since 4.1 no longer has fs._GridFS
784 chunks_collection = fs._GridFS__chunks if hasattr(fs, '_GridFS__chunks') else fs._chunks
785 files_collection = fs._GridFS__files if hasattr(fs, '_GridFS__files') else fs._files
786 ensure_index(chunks_collection, {'files_id': 1, 'n': 1}, unique=True)
787 ensure_index(files_collection, {'filename': 1, 'uploadDate': 1})
788
789 def sign(self, filter):
790 return signature(filter)
791
792 @staticmethod
793 def _cleanup(objrepr, tmppath=None):
794 # cleanup any temporary files on exit
795 # -- this is called as a finalizer (weakref.finalize)
796 try:
797 shutil.rmtree(tmppath, ignore_errors=True)
798 except Exception as e:
799 logger.error(f'finalizing {objrepr}: error occured as {e}')
800 return
801 try:
802 logger.debug(f'finalizing {objrepr}: cleaning up temporary files in {tmppath}')
803 except Exception:
804 pass