load_file.py
class Connection:
def __init__(self):
self.connection_type = None
self.server = None
self.port = None
self.user = None
self.password = None
self.isActive = None
self.file_extension = None
self.file_contains = None
self.file_location = None
self.schedule_minutes = None
self.interval_time = None
self.last_ran = None
def data_connection(self, data_connection_detail):
self.connection_type = data_connection_detail[1]
self.server = data_connection_detail[2]
self.port = data_connection_detail[3]
self.user = data_connection_detail[4]
self.password = data_connection_detail[5]
self.isActive = data_connection_detail[6]
self.file_extension = data_connection_detail[7]
self.file_contains = data_connection_detail[8]
self.file_location = data_connection_detail[9]
self.schedule_minutes = data_connection_detail[10]
self.interval_time = data_connection_detail[11]
self.last_ran = datetime.now()
def get_connection(self):
cursor = connection.cursor()
cursor.execute(
"SELECT ID, Type, Server, Port, User, Password, isActive, FileExtension, FileContains, FileLocation, "
"ScheduleMinutes, IntervalTime, LastRan from DataConnection WHERE isActive=True")
return cursor.fetchall()
server_ftp.py
class EasyFTP:
def __init__(self):
self.type_email = None
self.isActive = None
self.password = None
self.user = None
self.port = None
self.server = None
self.connection_type = None
self.file_extension = None
self.file_contains = None
self.file_location = None
def easy_ftp(self, data_connection_detail):
# enter FTP server fields into datagrip once active
self.type_email = data_connection_detail[1]
self.server = data_connection_detail[2]
self.port = data_connection_detail[3]
self.user = data_connection_detail[4]
self.password = data_connection_detail[5]
self.isActive = data_connection_detail[6]
self.file_extension = data_connection_detail[7]
self.file_contains = data_connection_detail[8]
self.file_location = data_connection_detail[9]
# using hard coded config for FTP server fields for testing
FTP_HOST = self.server
FTP_USER = self.user
FTP_PASS = self.password
# connect to the FTP server
with FTP(FTP_HOST, FTP_USER, FTP_PASS) as ftp:
ftp.encoding = "utf-8"
ftp.cwd('/files')
for filename in ftp.nlst():
# fnmatch compares single filename against a pattern '*' = matches everything
if fnmatch.fnmatch(filename, f'*{self.file_contains}*{self.file_extension}'):
with open(f'{self.file_location}/{filename}', 'wb') as fp:
ftp.retrbinary(f'RETR {filename}', fp.write)
我想做的是不要在easy_ftp函数中重复使用"data_connection_detail[1]、[2]等，因为这不是一个好的做法。相反，我尝试在easy_fp函数内部使用函数data_connection中的对象属性。
如果有什么我可以进一步澄清的，请告诉我。非常感谢。
部分解；这不是一个答案，也与这个问题无关，但这样的东西可能会更整洁一点，并使未来的修改更容易吗？
您当前有以下代码
def data_connection(self, data_connection_detail):
self.connection_type = data_connection_detail[1]
self.server = data_connection_detail[2]
self.port = data_connection_detail[3]
self.user = data_connection_detail[4]
self.password = data_connection_detail[5]
self.isActive = data_connection_detail[6]
self.file_extension = data_connection_detail[7]
self.file_contains = data_connection_detail[8]
self.file_location = data_connection_detail[9]
self.schedule_minutes = data_connection_detail[10]
self.interval_time = data_connection_detail[11]
self.last_ran = datetime.now()
这可以简化为更短的
def data_connection(self,data_connection_detail):
connection_type,server,port,user,password,isActive,file_extension,file_contains,file_location,schedule_minutes,interval_time,last_ran=*data_connection_detail[1:],datetime.now()
self.__dict__.update(locals())
或者，也许将data_connection_detail作为字典传递更容易？然后你可以做
def data_connection(self, data_connection):
self.__dict__.update(data_connection|{'last_ran':datetime.now()})
更好的做法是在EasyFTP类中继承Connection类，并调用父类的init(在本例中为Connection)，然后EasyFTP类将自动设置所有属性的
示例EasyFTP:
import Connection
class EasyFTP(Connection):
def __init__(self):
super().__init__()
print(self.server)