Did I find the right examples for you? yes no      Crawl my project      Python Jobs

All Samples(12)  |  Call(8)  |  Derive(0)  |  Import(4)
Local storage for stateful metadata and interprocess communication

This storage implementation is dummy and not scalable at all: loads all
metadata in memory and does frequent serialization / deserialization using
the Pickle module. Needs to be rewritten using sqlite and maybe SQLAlchemy
but it's a good enough model for prototyping the rest of the application and
identify the key requirements in terms of persistence and interprocess
communication.

src/n/u/nuxeo-otg-HEAD/notg/main.py   nuxeo-otg(Download)
import ConfigParser
 
from notg.storage import Storage
from notg.client import LocalClient, RemoteClient
from notg.synchronizer import Synchronizer
    remote_root = config.get("bindings", "remote")
 
    storage = Storage(os.path.join(home, "nuxeo.db"))
    local_client = LocalClient(local_root)
    remote_client = RemoteClient(server, username, password, remote_root)

src/n/u/nuxeo-otg-HEAD/notg/controller.py   nuxeo-otg(Download)
import os
import logging
 
from notg.storage import Storage
from notg.synchronizer import Synchronizer
        # where to persist the current bindings
        if storage is None or isinstance(storage, basestring):
            self.storage = Storage(storage)
        else:
            self.storage = storage

src/n/u/nuxeo-otg-HEAD/tests/test_synchronizer.py   nuxeo-otg(Download)
from cmislib.exceptions import ObjectNotFoundException
 
from notg.storage import Storage
from notg.client import LocalClient, RemoteClient
from notg.synchronizer import Synchronizer
        self.tempdir2 = tempfile.mkdtemp()
 
        self.storage = Storage(self.storagedir)
        self.local_client = LocalClient(self.tempdir1)
        self.remote_client = LocalClient(self.tempdir2)
    def setUp(self):
        self.storagedir = tempfile.mkdtemp()
        self.storage = Storage(self.storagedir)
 
        self.tempdir1 = tempfile.mkdtemp()

src/n/u/nuxeo-otg-HEAD/tests/test_storage.py   nuxeo-otg(Download)
from tempfile import mkdtemp
import unittest
import shutil
 
from notg.storage import Storage
    def setUp(self):
        # controller component to test
        self.storage_folder = mkdtemp('nuxeo-otg-storage-')
        self.storage = Storage(self.storage_folder)
 
        # reopen the storage from the disk and check that the binding is still
        # present
        storage = Storage(self.storage_folder)
        self.assertEquals(1, len(storage.list_bindings()))
 
 
        # open a new storage
        storage = Storage(self.storage_folder)
        self.assertEquals(1, len(storage.get_states(b)))
        s = storage.get_state(b, 'some_file.txt')
 
        # open a new storage and check that it's empty
        storage = Storage(self.storage_folder)
        self.assertEquals(0, len(storage.get_states(b)))