Commit 67ed55f3 authored by Cristiano Urban's avatar Cristiano Urban
Browse files

Code cleanup + added some doc. 



Signed-off-by: default avatarCristiano Urban <cristiano.urban@inaf.it>
parent 3bd34221

transfer_service/abort_job_amqp_server.py

+1 −6
Original line number Diff line number Diff line 
from amqp_server import AMQPServer

from db_connector import DbConnector

#from job_cache import JobCache

from config import Config
@@ -9,10 +8,6 @@ class AbortJobAMQPServer(AMQPServer): 
    def __init__(self, host, port, queue):

        self.type = "abort"

        config = Config("vos_ts.conf")        

        #self.params = config.loadSection("job_cache")

        #self.jobCache = JobCache(self.params["host"], 

        #                         self.params.getint("port"), 

        #                         self.params.getint("db_read"))

        self.params = config.loadSection("file_catalog")

        self.dbConn = DbConnector(self.params["user"], 

                                  self.params["password"],

transfer_service/checksum.py

+9 −2
Original line number Diff line number Diff line
@@ -13,15 +13,19 @@ class Checksum(object): 
        self.fileBufferSize = self.params.getint("file_buffer_size")

        self.md5FileSuffix = self.params["md5_file_suffix"]



    # Sets the buffer size in bytes when reading a chunk of data

    def setFileBufferSize(fileBufferSize):

        self.fileBufferSize = fileBufferSize

    

    # Checks whether a file is a checksum file or not

    def fileIsValid(self, absFilePath):

        if not self.md5FileSuffix in absFilePath:

            return True

        else:

            return False

     

    # Returns the MD5 checksum corresponding to a give filename

    # according to its absolute path on disk

    def getMD5(self, absFilePath):        

        if self.fileIsValid(absFilePath):

            md5FileName = os.path.dirname(absFilePath) + self.md5FileSuffix
@@ -33,6 +37,7 @@ class Checksum(object): 
                    return md5sum

        return None



    # Calculates the MD5 checksum of a file

    def md5sum(self, filePath):

        md5Hash = hashlib.md5()

        with open(filePath, "rb") as f:
@@ -40,6 +45,8 @@ class Checksum(object): 
                md5Hash.update(chunk)

        return md5Hash.hexdigest()



    # Calculates MD5 checksums recursively and saves on disk the list of

    # files in a folder with also the corresponding checksums

    def recursive(self, folderPath):

        for folder, subfolder, files in os.walk(folderPath, topdown = False):

            cwd = os.path.basename(folder)

transfer_service/get_job_amqp_server.py

+1 −9
Original line number Diff line number Diff line 
from amqp_server import AMQPServer

from db_connector import DbConnector

#from job_cache import JobCache

from config import Config
@@ -9,10 +8,6 @@ class GetJobAMQPServer(AMQPServer): 
    def __init__(self, host, port, queue):

        self.type = "poll"

        config = Config("vos_ts.conf")        

        #self.params = config.loadSection("job_cache")

        #self.jobCache = JobCache(self.params["host"], 

        #                         self.params.getint("port"), 

        #                         self.params.getint("db_read"))

        self.params = config.loadSection("file_catalog")

        self.dbConn = DbConnector(self.params["user"], 

                                  self.params["password"],
@@ -24,12 +19,9 @@ class GetJobAMQPServer(AMQPServer): 
    def execute_callback(self, requestBody):

        if "jobId" in requestBody:

            self.dbConn.connect()

            #redis_res = self.jobCache.get(requestBody["jobId"])

            dbResponse = self.dbConn.getJob(requestBody["jobId"])

            self.dbConn.disconnect()

            #print(f"Redis response: {redis_res}")

            print(f"Db response: {dbResponse}")

            #return redis_res

            return dbResponse

        else:

            return 42

transfer_service/start_job_amqp_server.py

+0 −10
Original line number Diff line number Diff line
@@ -4,7 +4,6 @@ import time # only for testing purposes 
from amqp_server import AMQPServer

from db_connector import DbConnector

from job import Job

#from job_cache import JobCache

from config import Config
@@ -13,10 +12,6 @@ class StartJobAMQPServer(AMQPServer): 
    def __init__(self, host, port, queue):

        self.type = "start"

        config = Config("vos_ts.conf")

        #self.params = config.loadSection("job_cache")

        #self.jobCache = JobCache(self.params["host"], 

        #                         self.params.getint("port"), 

        #                         self.params.getint("db_read"))

        self.params = config.loadSection("file_catalog")

        self.dbConn = DbConnector(self.params["user"], 

                                  self.params["password"],
@@ -32,16 +27,12 @@ class StartJobAMQPServer(AMQPServer): 
        self.job.setPhase("PENDING")

        self.dbConn.connect()

        self.job.setOwnerId("0000")

        #self.jobCache.set(self.job)

        #redis_res = self.jobCache.get(self.job.jobId)

        self.dbConn.insertJob(self.job)

        dbResponse = self.dbConn.getJob(self.job.jobId)

        self.dbConn.disconnect()

        #print(f"Redis response: {redis_res}")

        print(f"Db response: {dbResponse}")

        t = threading.Thread(target = self.fake_job) # only for testing purposes

        t.start()                                    # only for testing purposes

        #return redis_res

        return dbResponse

      

    def run(self):
@@ -54,7 +45,6 @@ class StartJobAMQPServer(AMQPServer): 
        print("fake_job: changing job state...")

        self.job.setPhase("EXECUTING")

        self.dbConn.connect()

        #self.jobCache.set(self.job)

        self.dbConn.insertJob(self.job)

        self.dbConn.disconnect()

        print("fake_job: state changed!")
 
 No newline at end of file

transfer_service/db_connector.py

+25 −25

File changed.

Contains only whitespace changes.