def __init__(self, db_path):
self.db_path = db_path
self.db = shelve.open(self.db_path, 'c')
self.lock = threading.Lock()
self.stopped = False
python类open()的实例源码
def __init__(self, filename, flag='c', protocol=None, writeback=False):
import anydbm
Shelf.__init__(self, anydbm.open(filename, flag), protocol, writeback)
def open(filename, flag='c', protocol=None, writeback=False):
"""Open a persistent dictionary for reading and writing.
The filename parameter is the base filename for the underlying
database. As a side-effect, an extension may be added to the
filename and more than one file may be created. The optional flag
parameter has the same interpretation as the flag parameter of
anydbm.open(). The optional protocol parameter specifies the
version of the pickle protocol (0, 1, or 2).
See the module's __doc__ string for an overview of the interface.
"""
return DbfilenameShelf(filename, flag, protocol, writeback)
def parse(self):
if not os.path.isdir("recipes"):
raise ParseError("No recipes directory found.")
self.__cache.open()
try:
self.__parse()
# config files overrule everything else
for c in self.__configFiles:
c = str(c) + ".yaml"
if not os.path.isfile(c):
raise ParseError("Config file {} does not exist!".format(c))
self.__parseUserConfig(c)
finally:
self.__cache.close()
def __generatePackages(self, nameFormatter, env, cacheKey, sandboxEnabled):
# use separate caches with and without sandbox
if sandboxEnabled:
cacheName = ".bob-packages-sb.pickle"
else:
cacheName = ".bob-packages.pickle"
# try to load the persisted packages
states = { n:s() for (n,s) in self.__states.items() }
rootPkg = Package()
rootPkg.construct("<root>", [], nameFormatter, None, [], [], states,
{}, {}, None, None, [], {}, -1)
try:
with open(cacheName, "rb") as f:
persistedCacheKey = f.read(len(cacheKey))
if cacheKey == persistedCacheKey:
tmp = PackageUnpickler(f, self.getRecipe, self.__plugins,
nameFormatter).load()
return tmp.toStep(nameFormatter, rootPkg).getPackage()
except (EOFError, OSError, pickle.UnpicklingError):
pass
# not cached -> calculate packages
result = self.__rootRecipe.prepare(nameFormatter, env, sandboxEnabled,
states)[0]
# save package tree for next invocation
tmp = CoreStepRef(rootPkg, result.getPackageStep())
try:
newCacheName = cacheName + ".new"
with open(newCacheName, "wb") as f:
f.write(cacheKey)
PackagePickler(f, nameFormatter).dump(tmp)
os.replace(newCacheName, cacheName)
except OSError as e:
print("Error saving internal state:", str(e), file=sys.stderr)
return result
def open(self):
self.__shelve = shelve.open(".bob-cache.shelve")
self.__files = {}
def loadYaml(self, name, yamlSchema, default):
bs = binStat(name)
if name in self.__shelve:
cached = self.__shelve[name]
if ((cached['lstat'] == bs) and
(cached.get('vsn') == BOB_INPUT_HASH)):
self.__files[name] = cached['digest']
return cached['data']
with open(name, "r") as f:
try:
rawData = f.read()
data = yaml.safe_load(rawData)
digest = hashlib.sha1(rawData.encode('utf8')).digest()
except Exception as e:
raise ParseError("Error while parsing {}: {}".format(name, str(e)))
if data is None: data = default
try:
data = yamlSchema.validate(data)
except schema.SchemaError as e:
raise ParseError("Error while validating {}: {}".format(name, str(e)))
self.__files[name] = digest
self.__shelve[name] = {
'lstat' : bs,
'data' : data,
'vsn' : BOB_INPUT_HASH,
'digest' : digest
}
return data
def loadBinary(self, name):
with open(name, "rb") as f:
result = f.read()
self.__files[name] = hashlib.sha1(result).digest()
return result
def build_prj1_code_graph(self):
if not self.prj1_scitools_client.project_exists():
print('understand project does not exist, '
'first run "$ prj1 understand --build"')
else:
with shelve.open(str(self.shelve_prj1_code_db_path)) as db:
self.prj1_scitools_client.open_project()
scitools_project = self.prj1_scitools_client.build_project(
self.prj1_code_repo_path)
self.prj1_scitools_client.close_project()
db['code_graph'] = scitools_project
print('loaded scitools project of size',
len(scitools_project.code_graph))
print('entity kinds:', scitools_project.entity_kinds)
print('ref kinds:', scitools_project.ref_kinds)
def build_git_graph(self):
with shelve.open(str(self.shelve_db_path)) as db:
git_client = GitClient(self.git_repo_path)
git_graph = git_client.build_commit_graph()
git_client.add_commit_tree(git_graph, ref_name='origin/master')
db['git_graph'] = git_graph
self.git_graph = git_graph
def load_git_graph(self):
with shelve.open(str(self.shelve_db_path)) as db:
if 'git_graph' in db:
self.git_graph = db['git_graph']
else:
self.git_graph = None
return self.git_graph
def build_code_graph(self):
if not self.scitools_client.project_exists():
print('understand project does not exist, '
'first run "$ povray understand --build"')
else:
with shelve.open(str(self.shelve_db_path)) as db:
self.scitools_client.open_project()
self.scitools_project = self.scitools_client.build_project(
self.git_repo_path)
self.scitools_client.close_project()
db['code_graph'] = self.scitools_project
print('loaded scitools project of size',
len(self.scitools_project.code_graph))
print('entity kinds:', self.scitools_project.entity_kinds)
print('ref kinds:', self.scitools_project.entity_kinds)
def load_code_graph(self) -> ScitoolsProject:
with shelve.open(str(self.shelve_db_path)) as db:
if 'code_graph' in db:
self.scitools_project = db['code_graph']
else:
self.scitools_project = None
return self.scitools_project
def test_unshelve_similarity_graph(data_root):
with shelve.open(str(data_root / 'test-similarity_graph.shelve')) as db:
graph = db['similarity_graph']
assert graph is not None
print()
actor_groups = graph.group_similar_actors()
pprint(actor_groups)
def load_game(mySaveNum):
myShelfFile = "./save"+str(mySaveNum)
print("Save file: {0}".format(myShelfFile))
myOpenShelf = shelve.open(myShelfFile)
cities = myOpenShelf['cities']
clubs = myOpenShelf['clubs']
humans = myOpenShelf['humans']
leagues = myOpenShelf['leagues']
managers = myOpenShelf['managers']
chairmans = myOpenShelf['chairmans']
currentCity = myOpenShelf['currentCity']
chosenClub = myOpenShelf['chosenClub']
chosenLeague = myOpenShelf['chosenLeague']
myOpenShelf.close()
menuMain(chosenLeague, chosenClub)
def save_game(myChosenLeague, myChosenClub, mySaveNum=0, myCurrentCity=0):
myShelfFile = "./save"+str(mySaveNum)
print("Save file: {0}".format(myShelfFile))
myOpenShelf = shelve.open(myShelfFile)
myOpenShelf['cities'] = cities
myOpenShelf['clubs'] = clubs
myOpenShelf['humans'] = humans
myOpenShelf['leagues'] = leagues
myOpenShelf['managers'] = managers
myOpenShelf['chairmans'] = chairmans
myOpenShelf['currentCity'] = myCurrentCity
myOpenShelf['chosenClub'] = myChosenClub
myOpenShelf['chosenLeague'] = myChosenLeague
myOpenShelf.close()
def get(self, *a, **kw):
self.open()
return super(FileCache, self).get(*a, **kw)
def save(self):
self.open()
self.storage.sync()
logger.debug("Saved %s.", SETTINGS.CACHE_PATH)
