2816

pending completion

Build # 2816

Build Type

Pull #332

travis-ci

Committed by

web-flow

Commit Message

Revert "Removed text encoding from filepad.py, replaced by raw access in order to be able to store both text and binary files in Filepad"

This reverts commit 010b6fbbf.

Pull Request Pull Request #332: GetFilesByQueryTask suggestion

Run Details

38 of 38 new or added lines in 2 files covered. (100.0%)

3256 of 5392 relevant lines covered (60.39%)

1.21 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

64.56

/fireworks/user_objects/firetasks/filepad_tasks.py

# coding: utf-8

from __future__ import unicode_literals

import os

from fireworks.core.firework import FiretaskBase
from fireworks.utilities.filepad import FilePad

__author__ = 'Kiran Mathew'
__email__ = 'kmathew@lbl.gov'
__credits__ = 'Anubhav Jain'


class AddFilesTask(FiretaskBase):
    """
    A Firetask to add files to the filepad.

    Required params:
        - paths (list/str): either list of paths or a glob pattern string.

    Optional params:
        - identifiers ([str]): list of identifiers, one for each file in the paths list
        - directory (str): path to directory where the pattern matching is to be done.
        - filepad_file (str): path to the filepad db config file
        - compress (bool): whether or not to compress the file before inserting to gridfs
        - metadata (dict): metadata to store along with the file, stored in 'metadata' key
    """
    _fw_name = 'AddFilesTask'
    required_params = ["paths"]
    optional_params = ["identifiers", "directory", "filepad_file", "compress", "metadata"]

    def run_task(self, fw_spec):

        from glob import glob

        directory = os.path.abspath(self.get("directory", "."))

        if isinstance(self["paths"], list):
            paths = [os.path.abspath(p) for p in self["paths"]]
        else:
            paths = [os.path.abspath(p) for p in glob("{}/{}".format(directory, self["paths"]))]

        # if not given, use the full paths as identifiers
        identifiers = self.get("identifiers", paths)

        assert len(paths) == len(identifiers)

        fpad = get_fpad(self.get("filepad_file", None))

        for p, l in zip(paths, identifiers):
            fpad.add_file(p, identifier=l, metadata=self.get("metadata", None),
                          compress=self.get("compress", True))


class GetFilesTask(FiretaskBase):
    """
    A Firetask to fetch files from the filepad and write it to specified directory(current working
    directory if not specified)

    Required params:
        - identifiers ([str]): identifiers of files to fetch

    Optional params:
        - filepad_file (str): path to the filepad db config file
        - dest_dir (str): destination directory, default is the current working directory
        - new_file_names ([str]): if provided, the retrieved files will be renamed
    """
    _fw_name = 'GetFilesTask'
    required_params = ["identifiers"]
    optional_params = ["filepad_file", "dest_dir", "new_file_names"]

    def run_task(self, fw_spec):
        fpad = get_fpad(self.get("filepad_file", None))
        dest_dir = self.get("dest_dir", os.path.abspath("."))
        new_file_names = self.get("new_file_names", [])
        for i, l in enumerate(self["identifiers"]):
            file_contents, doc = fpad.get_file(identifier=l)
            file_name = new_file_names[i] if new_file_names else doc["original_file_name"]
            with open(os.path.join(dest_dir, file_name), "w") as f:
                f.write(file_contents.decode())

class GetFilesByQueryTask(FiretaskBase):
    """
    A Firetask to fetch files from the filepad and write it to specified directory (current working
    directory if not specified)

    Required params:
        - query ([str]): mongo db query identifying files to fetch

    Optional params:
        - sort_key (str): sort key, don't sort per default
        - sort_direction (int): sort direction, default 'pymongo-DESCENDING'
        - limit (int): maximum number of files to write, default: no limit
        - filepad_file (str): path to the filepad db config file
        - dest_dir (str): destination directory, default is the current working
          directory.
        - new_file_names ([str]): if provided, the retrieved files will be
          renamed. Not recommended as order and number of queried files not fixed.
    """
    _fw_name = 'GetFilesByQueryTask'
    required_params = ["query"]
    optional_params = ["sort_key","sort_direction", "limit",
        "filepad_file", "dest_dir", "new_file_names"]

    def run_task(self, fw_spec):
        import pymongo
        from pprint import pprint
        fpad = get_fpad(self.get("filepad_file", None))
        dest_dir = self.get("dest_dir", os.path.abspath("."))
        new_file_names = self.get("new_file_names", [])
        query = self.get("query", {})
        sort_key = self.get("query", None)
        sort_direction = self.get("query", pymongo.DESCENDING)
        limit = self.get("limit",None)

        # convert nested dicts to MongoDB query, i.e.
        # { 'metadata' : { 'key': 'value' } --> { 'metadata.key': 'value' }
        # ATTENTION: this does not work for more sophisticated queries
        # such as { 'identifier': { '$in': [ 'id1', 'id2' ] } }
        # TODO: find proper query translation method
        def nested2plain(d, prefix=''):
            r = {}
            if prefix is not '':
                prefix = prefix + '.'
            for k, v in d.items():
                if type(v) is dict:
                    r.update( nested2plain(v, prefix + k) )
                else:
                    r.update( {prefix + k: v} )
            return r

        query = nested2plain(query)
        pprint(query)

        l = fpad.get_file_by_query(query,sort_key,sort_direction)
        for i, (file_contents, doc) in enumerate(l):
            if isinstance(limit, int) and i >= limit:
                break # maximum number of files reached
            file_name = new_file_names[i] if new_file_names else doc["original_file_name"]
            with open(os.path.join(dest_dir, file_name), "wb") as f:
                f.write(file_contents)

class DeleteFilesTask(FiretaskBase):
    """
    A Firetask to delete files from the filepad

    Required params:
        - identifiers ([str]): identifiers of files to delete

    Optional params:
        - filepad_file (str): path to the filepad db config file
    """
    _fw_name = 'DeleteFilesTask'
    required_params = ["identifiers"]
    optional_params = ["filepad_file"]

    def run_task(self, fw_spec):
        fpad = get_fpad(self.get("filepad_file", None))
        for l in self["identifiers"]:
            fpad.delete_file(l)


def get_fpad(fpad_file):
    if fpad_file:
        return FilePad.from_db_file(fpad_file)
    else:
        return FilePad.auto_load()

1	# coding: utf-8
2
3	from __future__ import unicode_literals	2✔
4
5	import os	2✔
6
7	from fireworks.core.firework import FiretaskBase	2✔
8	from fireworks.utilities.filepad import FilePad	2✔
9
10	__author__ = 'Kiran Mathew'	2✔
11	__email__ = 'kmathew@lbl.gov'	2✔
12	__credits__ = 'Anubhav Jain'	2✔
13
14
15	class AddFilesTask(FiretaskBase):	2✔
16	"""
17	A Firetask to add files to the filepad.
18
19	Required params:
20	- paths (list/str): either list of paths or a glob pattern string.
21
22	Optional params:
23	- identifiers ([str]): list of identifiers, one for each file in the paths list
24	- directory (str): path to directory where the pattern matching is to be done.
25	- filepad_file (str): path to the filepad db config file
26	- compress (bool): whether or not to compress the file before inserting to gridfs
27	- metadata (dict): metadata to store along with the file, stored in 'metadata' key
28	"""
29	_fw_name = 'AddFilesTask'	2✔
30	required_params = ["paths"]	2✔
31	optional_params = ["identifiers", "directory", "filepad_file", "compress", "metadata"]	2✔
32
33	def run_task(self, fw_spec):	2✔
34
35	from glob import glob	2✔
36
37	directory = os.path.abspath(self.get("directory", "."))	2✔
38
39	if isinstance(self["paths"], list):	2✔
40	paths = [os.path.abspath(p) for p in self["paths"]]	2✔
41	else:
42	paths = [os.path.abspath(p) for p in glob("{}/{}".format(directory, self["paths"]))]	2✔
43
44	# if not given, use the full paths as identifiers
45	identifiers = self.get("identifiers", paths)	2✔
46
47	assert len(paths) == len(identifiers)	2✔
48
49	fpad = get_fpad(self.get("filepad_file", None))	2✔
50
51	for p, l in zip(paths, identifiers):	2✔
52	fpad.add_file(p, identifier=l, metadata=self.get("metadata", None),	2✔
53	compress=self.get("compress", True))
54
55
56	class GetFilesTask(FiretaskBase):	2✔
57	"""
58	A Firetask to fetch files from the filepad and write it to specified directory(current working
59	directory if not specified)
60
61	Required params:
62	- identifiers ([str]): identifiers of files to fetch
63
64	Optional params:
65	- filepad_file (str): path to the filepad db config file
66	- dest_dir (str): destination directory, default is the current working directory
67	- new_file_names ([str]): if provided, the retrieved files will be renamed
68	"""
69	_fw_name = 'GetFilesTask'	2✔
70	required_params = ["identifiers"]	2✔
71	optional_params = ["filepad_file", "dest_dir", "new_file_names"]	2✔
72
73	def run_task(self, fw_spec):	2✔
74	fpad = get_fpad(self.get("filepad_file", None))	2✔
75	dest_dir = self.get("dest_dir", os.path.abspath("."))	2✔
76	new_file_names = self.get("new_file_names", [])	2✔
77	for i, l in enumerate(self["identifiers"]):	2✔
78	file_contents, doc = fpad.get_file(identifier=l)	2✔
79	file_name = new_file_names[i] if new_file_names else doc["original_file_name"]	2✔
80	with open(os.path.join(dest_dir, file_name), "w") as f:	2✔
81	f.write(file_contents.decode())	2✔
82
83	class GetFilesByQueryTask(FiretaskBase):	2✔
84	"""
85	A Firetask to fetch files from the filepad and write it to specified directory (current working
86	directory if not specified)
87
88	Required params:
89	- query ([str]): mongo db query identifying files to fetch
90
91	Optional params:
92	- sort_key (str): sort key, don't sort per default
93	- sort_direction (int): sort direction, default 'pymongo-DESCENDING'
94	- limit (int): maximum number of files to write, default: no limit
95	- filepad_file (str): path to the filepad db config file
96	- dest_dir (str): destination directory, default is the current working
97	directory.
98	- new_file_names ([str]): if provided, the retrieved files will be
99	renamed. Not recommended as order and number of queried files not fixed.
100	"""
101	_fw_name = 'GetFilesByQueryTask'	2✔
102	required_params = ["query"]	2✔
103	optional_params = ["sort_key","sort_direction", "limit",	2✔
104	"filepad_file", "dest_dir", "new_file_names"]
105
106	def run_task(self, fw_spec):	2✔
107	import pymongo	×
108	from pprint import pprint	×
109	fpad = get_fpad(self.get("filepad_file", None))	×
110	dest_dir = self.get("dest_dir", os.path.abspath("."))	×
111	new_file_names = self.get("new_file_names", [])	×
112	query = self.get("query", {})	×
113	sort_key = self.get("query", None)	×
114	sort_direction = self.get("query", pymongo.DESCENDING)	×
115	limit = self.get("limit",None)	×
116
117	# convert nested dicts to MongoDB query, i.e.
118	# { 'metadata' : { 'key': 'value' } --> { 'metadata.key': 'value' }
119	# ATTENTION: this does not work for more sophisticated queries
120	# such as { 'identifier': { '$in': [ 'id1', 'id2' ] } }
121	# TODO: find proper query translation method
122	def nested2plain(d, prefix=''):	×
123	r = {}	×
124	if prefix is not '':	×
125	prefix = prefix + '.'	×
126	for k, v in d.items():	×
127	if type(v) is dict:	×
128	r.update( nested2plain(v, prefix + k) )	×
129	else:
130	r.update( {prefix + k: v} )	×
131	return r	×
132
133	query = nested2plain(query)	×
134	pprint(query)	×
135
136	l = fpad.get_file_by_query(query,sort_key,sort_direction)	×
137	for i, (file_contents, doc) in enumerate(l):	×
138	if isinstance(limit, int) and i >= limit:	×
139	break # maximum number of files reached	×
140	file_name = new_file_names[i] if new_file_names else doc["original_file_name"]	×
141	with open(os.path.join(dest_dir, file_name), "wb") as f:	×
142	f.write(file_contents)	×
143
144	class DeleteFilesTask(FiretaskBase):	2✔
145	"""
146	A Firetask to delete files from the filepad
147
148	Required params:
149	- identifiers ([str]): identifiers of files to delete
150
151	Optional params:
152	- filepad_file (str): path to the filepad db config file
153	"""
154	_fw_name = 'DeleteFilesTask'	2✔
155	required_params = ["identifiers"]	2✔
156	optional_params = ["filepad_file"]	2✔
157
158	def run_task(self, fw_spec):	2✔
159	fpad = get_fpad(self.get("filepad_file", None))	2✔
160	for l in self["identifiers"]:	2✔
161	fpad.delete_file(l)	2✔
162
163
164	def get_fpad(fpad_file):	2✔
165	if fpad_file:	2✔
166	return FilePad.from_db_file(fpad_file)	×
167	else:
168	return FilePad.auto_load()	2✔

materialsproject / fireworks / 2816

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous