Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
@uploader.ensure_prepared
def load_data(self,data_folder):
# there's one zip there, let's get the zipped filename
zgrasp = glob.glob(os.path.join(data_folder,"*.zip"))
if len(zgrasp) != 1:
raise uploader.ResourceError("Expecting one zip only, got: %s" % repr(zgrasp))
zgrasp = zgrasp.pop()
zf = zipfile.ZipFile(zgrasp)
content = [e.filename for e in zf.filelist]
if len(content) != 1:
raise uploader.ResourceError("Expecting only one file in the archive, got: %s" % content)
input_file = content.pop()
input_file = os.path.join(data_folder,input_file)
self.logger.info("Load data from file '%s'" % input_file)
return load_data(input_file)
import os
import glob
import zipfile
from .emv_parser import load_data
import biothings.dataload.uploader as uploader
class EMVUploader(uploader.BaseSourceUploader):
name = "emv"
@uploader.ensure_prepared
def load_data(self,data_folder):
# there's one csv file there, let's get it
input_file = glob.glob(os.path.join(data_folder,"EmVClass*.csv"))
if len(input_file) != 1:
raise uploader.ResourceError("Expecting only one CSV file, got: %s" % input_file)
input_file = input_file.pop()
self.logger.info("Load data from file '%s'" % input_file)
return load_data(input_file)
@classmethod
def get_mapping(klass):
from .ucsc_base import load_ucsc_exons
import biothings.dataload.uploader as uploader
class UCSCExonsUploader(uploader.MergerSourceUploader):
name = "ucsc_exons"
main_source = "ucsc"
def load_data(self,data_folder):
genedoc_d = load_ucsc_exons(data_folder)
return genedoc_d
def get_mapping(self):
mapping = {
#do not index exons
"exons": {
"type": "object",
"enabled": False
},
#do not index exons_hg19
def load_data(self,data_folder):
# there's one csv file there, let's get it
input_file = glob.glob(os.path.join(data_folder,"EmVClass*.csv"))
if len(input_file) != 1:
raise uploader.ResourceError("Expecting only one CSV file, got: %s" % input_file)
input_file = input_file.pop()
self.logger.info("Load data from file '%s'" % input_file)
return load_data(input_file)
from .clinvar_xml_parser import load_data as load_common
from . import get_mapping
import biothings.dataload.uploader as uploader
class ClinvarHG38Uploader(uploader.BaseSourceUploader):
name = "clinvar_hg38"
main_source = "clinvar"
def load_data(self,data_folder):
self.logger.info("Load data from folder '%s'" % data_folder)
return load_common(self,hg19=False,data_folder=data_folder)
def get_mapping(self):
return get_mapping()
@uploader.ensure_prepared
def load_data(self,data_folder):
self.logger.info("Load data from folder '%s'" % data_folder)
return load_data(data_folder)
from .cadd_parser import load_data
import biothings.dataload.uploader as uploader
class CADDUploader(uploader.BaseSourceUploader):
name = "cadd"
@uploader.ensure_prepared
def load_data(self,data_folder):
self.logger.info("Load data from folder '%s'" % data_folder)
return load_data(data_folder)
@classmethod
def get_mapping(klass):
mapping = {
"cadd": {
"properties": {
"annotype": {
"type": "string",
"analyzer": "string_lowercase"
from .uniprot_base import load_pir
import biothings.dataload.uploader as uploader
class UniprotPIRUploader(uploader.MergerSourceUploader):
name = "uniprot_pir"
main_source = "uniprot"
def load_data(self, data_folder):
return load_pir(data_folder)
def get_mapping(self):
mapping = {
"pir": {
"type": "string",
"analyzer": "string_lowercase"
}
}
return mapping
import os
import glob
import zipfile
from .grasp_parser import load_data
import biothings.dataload.uploader as uploader
class GraspUploader(uploader.BaseSourceUploader):
name = "grasp"
@uploader.ensure_prepared
def load_data(self,data_folder):
# there's one zip there, let's get the zipped filename
zgrasp = glob.glob(os.path.join(data_folder,"*.zip"))
if len(zgrasp) != 1:
raise uploader.ResourceError("Expecting one zip only, got: %s" % repr(zgrasp))
zgrasp = zgrasp.pop()
zf = zipfile.ZipFile(zgrasp)
content = [e.filename for e in zf.filelist]
if len(content) != 1:
raise uploader.ResourceError("Expecting only one file in the archive, got: %s" % content)
input_file = content.pop()
input_file = os.path.join(data_folder,input_file)