def remove_summaries_by_eids(self, eids):
f_name = self.items["result log"]["db_file"]
with TinyDB(f_name) as db_entity:
return db_entity.remove(eids=eids)
python类TinyDB()的实例源码
def __init__(self):
"""Adds self to messages and event's `data` field.
Through this instance you can access TinyDB instance (data["tinydbproxy"].tinydb).
This plugin should be included first!
"""
super().__init__()
self.tinydb = TinyDB(path=self.get_path("tinydb_database.json"), storage=CachingMiddleware(JSONStorage))
def _opendb(self):
self.middleware = CachingMiddleware(JSONStorage)
self.middleware.WRITE_CACHE_SIZE = 500
self.db = TinyDB(self.conn_str, storage=self.middleware,
default_table=self.default_table)
def main():
"""Method to put it all together"""
db = TinyDB(os.path.join(os.getcwd(), DATABASE_NAME))
for sort_type in ["hot", "new", "top"]:
logger.debug("Fetching url: {}".format(sort_type))
json_data = get_json_from_subreddit(sort_type, 100)
json_list = []
for data in json_data:
for post in data['data']['children']:
if is_post_valid(post):
j = convert_post_to_json(post)
if j is not None:
insert_into_database(db, j)
def get_genres(database_name):
"""Utility method to get all the genres as a set"""
db = TinyDB(os.path.join(os.getcwd(), database_name))
all_genres = { song['genre'] for song in db.all() }
specific_genres = set()
for genre in all_genres:
specific_genres = specific_genres.union(set(genre.strip().split('/')))
db.close()
return _strip_spaces(specific_genres)
def get_total_songs(database_name):
"""Utility Method to get the total number of songs in the database"""
db = TinyDB(os.path.join(os.getcwd(), database_name))
total_length = len(db.all())
db.close()
return total_length
def setUp(self):
with open(os.path.join(BASE_DIR, 'tests', 'fixtures', 'test.json')) as f:
self.j = json.load(f)
self.database_name = os.path.join(os.getcwd(), 'test.db')
db = TinyDB(self.database_name)
db.insert_multiple(self.j)
db.close
def __init__(self, page=None, db=None):
self.db = TinyDB(db)
self.page = page
# TODO need to do some check that database is well formed
# TODO and that the page is a well-formed WebPage object
self.db.insert({
'url': self.page.link,
'title': self.page.title,
'text': self.page.text,
'links': self.page.links})
def gatherText(self):
''' Takes a TinyDB with a "text" field and combines all the text into
a one list of texts
'''
all_text = ""
for item in self.db:
all_text = all_text + " " + clean_text(item['text'])
self.text = all_text
def set(self, name, value):
db = TinyDB(self.filename)
db.upsert({'name': name, 'value': value}, where('name') == name)
def get(self, name, default=None):
db = TinyDB(self.filename)
item = db.get(where('name') == name)
if item is not None:
return item.get('value', default)
return default
def __init__(self):
self.node_details_content = None
self.public_files_db = TinyDB('public_files.json')
self.table = self.public_files_db.table('public_files')
def __init__(self):
self.ownstorj_playlists_db = TinyDB('ownstorj_playlists.json')
self.playlists_table = self.ownstorj_playlists_db.table('playlists')
self.tracks_table = self.ownstorj_playlists_db.table('tracks')
def __init__(self, **attributes):
self.__dict__ = attributes
self.db = TinyDB(TINY_DB_PATH)
self.table = self.db.table(self.__class__.__name__)
def list_log_services():
projectId = TinyDB('projects.json').table("Project").all()
resp, content = storage.get().authorize(Http()).request("https://logging.googleapis.com/v1beta3/projects/" + projectId + "/logServices","GET")
return [service['name'] for service in json.loads(content)['logServices']]
def insert_templates():
projectId = TinyDB('projects.json').table("Project").all()
request = instanceTemplates.list(project=projectId)
try:
while request is not None:
response = request.execute()
for instanceTemplate in response['items']:
template_table.insert(instanceTemplate)
request = instanceTemplates.list_next(previous_request=request, previous_response=response)
except KeyError:
pass
def insert_instance_groups():
projectId = TinyDB('projects.json').table("Project").all()
for zone in get_zones():
request = instanceGroups.list(project=projectId, zone=zone)
try:
while request is not None:
response = request.execute()
for instanceGroup in response['items']:
group_table.insert(instanceGroup)
request = instanceGroups.list_next(previous_request=request, previous_response=response)
except KeyError:
pass
def get_zones():
projectId = TinyDB('projects.json').table("Project").all()
results = []
request = zones.list(project=projectId)
response = request.execute()['items']
for result in response:
results.append(result['name'])
return results
def loadDBs(service, indexFiles):
dBs = {}
datadir = get_data_directory(service)
indexMetadata = getIndexMetadata(service)
#Files in Lambda can only be created in the /tmp filesystem - If it doesn't exist, create it.
lambdaFileSystem = '/tmp/'+service+'/data'
if not os.path.exists(lambdaFileSystem):
os.makedirs(lambdaFileSystem)
for i in indexFiles:
db = tinydb.TinyDB(lambdaFileSystem+'/'+i+'.json')
#TODO: remove circular dependency from utils, so I can use the method get_index_file_name
#TODO: initial tests show that is faster (by a few milliseconds) to populate the file from scratch). See if I should load from scratch all the time
#TODO:Create a file that is an index of those files that have been generated, so the code knows which files to look for and avoid creating unnecesary empty .json files
if len(db) == 0:
try:
with open(datadir+i+'.csv', 'rb') as csvfile:
pricelist = csv.DictReader(csvfile, delimiter=',', quotechar='"')
db.insert_multiple(pricelist)
except IOError:
pass
dBs[i]=db
return dBs, indexMetadata
def __init__(self):
self.db = TinyDB(os.path.join(os.path.join(os.getcwd(), os.path.dirname(__file__)), ListCache.DB_FILE))
