Skip to content

refactor: fix too-many-statements and too-many-branches #3437

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
gbrlcoutinho wants to merge 1 commit into
base: master
Choose a base branch
from
Open

refactor: fix too-many-statements and too-many-branches #3437

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
30 changes: 19 additions & 11 deletions luigi/batch_notifier.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -54,6 +54,14 @@ def _plural_format(template, number, plural="s"):
return template.format(number, "" if number == 1 else plural)


class _EmailConfig:
__slots__ = ("format", "default_owner")

def __init__(self, fmt, default_owner):
self.format = fmt
self.default_owner = default_owner


class BatchNotifier:
def __init__(self, **kwargs):
self._config = batch_email(**kwargs)
Expand All @@ -63,11 +71,11 @@ def __init__(self, **kwargs):
self._fail_expls = collections.defaultdict(_fail_queue(self._config.error_messages))
self._update_next_send()

self._email_format = email().format
if email().receiver:
self._default_owner = set(filter(None, email().receiver.split(",")))
else:
self._default_owner = set()
_email = email()
self._email_config = _EmailConfig(
fmt=_email.format,
default_owner=set(filter(None, _email.receiver.split(","))) if _email.receiver else set(),
)

def _update_next_send(self):
self._next_send = time.time() + 60 * self._config.email_interval
Expand All @@ -85,14 +93,14 @@ def _key(self, task_name, family, unbatched_args):

def _format_expl(self, expl):
lines = expl.rstrip().split("\n")[-self._config.error_lines :]
if self._email_format == "html":
if self._email_config.format == "html":
return "<pre>{}</pre>".format("\n".join(lines))
else:
return "\n{}".format("\n".join(map(" {}".format, lines)))

def _expl_body(self, expls):
lines = [self._format_expl(expl) for expl in expls]
if lines and self._email_format != "html":
if lines and self._email_config.format != "html":
lines.append("")
return "\n".join(lines)

Expand All @@ -108,13 +116,13 @@ def _format_task(self, task_tuple):

def _format_tasks(self, tasks):
lines = map(self._format_task, sorted(tasks, key=self._expl_key))
if self._email_format == "html":
if self._email_config.format == "html":
return "<li>{}".format("\n<br>".join(lines))
else:
return "- {}".format("\n ".join(lines))

def _owners(self, owners):
return self._default_owner | set(owners)
return self._email_config.default_owner | set(owners)

def add_failure(self, task_name, family, unbatched_args, expl, owners):
key = self._key(task_name, family, unbatched_args)
Expand Down Expand Up @@ -164,7 +172,7 @@ def _email_body(self, fail_counts, disable_counts, scheduling_counts, fail_expls
body_lines.append(self._format_tasks(tasks))
body_lines.append(msg)
body = "\n".join(filter(None, body_lines)).rstrip()
if self._email_format == "html":
if self._email_config.format == "html":
return "<ul>\n{}\n</ul>".format(body)
else:
return body
Expand All @@ -180,7 +188,7 @@ def _send_email(self, fail_counts, disable_counts, scheduling_counts, fail_expls
]
subject_base = ", ".join(filter(None, subject_parts))
if subject_base:
prefix = "" if owner in self._default_owner else "Your tasks have "
prefix = "" if owner in self._email_config.default_owner else "Your tasks have "
subject = "Luigi: {}{} in the last {} minutes".format(prefix, subject_base, self._config.email_interval)
email_body = self._email_body(fail_counts, disable_counts, scheduling_counts, fail_expls)
send_email(subject, email_body, email().sender, (owner,))
Expand Down
17 changes: 11 additions & 6 deletions luigi/contrib/azureblob.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -198,23 +198,28 @@ def splitfilepath(filepath):
return container, blob


class ReadableAzureBlobFile:
class _ReadableAzureBlobConfig:
def __init__(self, container, blob, client, download_when_reading, **kwargs):
self.container = container
self.blob = blob
self.client = client
self.closed = False
self.download_when_reading = download_when_reading
self.azure_blob_options = kwargs


class ReadableAzureBlobFile:
def __init__(self, container, blob, client, download_when_reading, **kwargs):
self._config = _ReadableAzureBlobConfig(container, blob, client, download_when_reading, **kwargs)
self.closed = False
self.download_file_location = os.path.join(tempfile.mkdtemp(prefix=str(datetime.datetime.utcnow())), blob)
self.fid = None

def read(self, n=None):
return self.client.download_as_bytes(self.container, self.blob, n)
return self._config.client.download_as_bytes(self._config.container, self._config.blob, n)

def __enter__(self):
if self.download_when_reading:
self.client.download_as_file(self.container, self.blob, self.download_file_location)
if self._config.download_when_reading:
self._config.client.download_as_file(self._config.container, self._config.blob, self.download_file_location)
self.fid = open(self.download_file_location)
return self.fid
else:
Expand All @@ -229,7 +234,7 @@ def __del__(self):
os.remove(self.download_file_location)

def close(self):
if self.download_when_reading:
if self._config.download_when_reading:
if self.fid is not None and not self.fid.closed:
self.fid.close()
self.fid = None
Expand Down
19 changes: 9 additions & 10 deletions luigi/contrib/esindex.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -87,6 +87,7 @@ def docs(self):
import itertools
import json
import logging
from collections import namedtuple

import luigi

Expand All @@ -103,6 +104,8 @@ def docs(self):
except ImportError:
logger.warning("Loading esindex module without elasticsearch installed. Will crash at runtime if esindex functionality is used.")

_ConnectionConfig = namedtuple('_ConnectionConfig', ['host', 'port', 'http_auth', 'timeout', 'extra_elasticsearch_args'])


class ElasticsearchTarget(luigi.Target):
"""Target for a resource in Elasticsearch."""
Expand Down Expand Up @@ -132,23 +135,19 @@ def __init__(self, host, port, index, doc_type, update_id, marker_index_hist_siz
if extra_elasticsearch_args is None:
extra_elasticsearch_args = {}

self.host = host
self.port = port
self.http_auth = http_auth
self._connection_config = _ConnectionConfig(host, port, http_auth, timeout, extra_elasticsearch_args)
self.index = index
self.doc_type = doc_type
self.update_id = update_id
self.marker_index_hist_size = marker_index_hist_size
self.timeout = timeout
self.extra_elasticsearch_args = extra_elasticsearch_args

self.es = elasticsearch.Elasticsearch(
connection_class=Urllib3HttpConnection,
host=self.host,
port=self.port,
http_auth=self.http_auth,
timeout=self.timeout,
**self.extra_elasticsearch_args,
host=self._connection_config.host,
port=self._connection_config.port,
http_auth=self._connection_config.http_auth,
timeout=self._connection_config.timeout,
**self._connection_config.extra_elasticsearch_args,
)

def marker_index_document_id(self):
Expand Down
60 changes: 30 additions & 30 deletions luigi/contrib/ftp.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -37,34 +37,36 @@
import luigi.format
import luigi.local_target
import luigi.target
from collections import namedtuple
from luigi.format import FileWrapper

logger = logging.getLogger("luigi-interface")


_RemoteFileSystemConfig = namedtuple('_RemoteFileSystemConfig', [
'host', 'username', 'password', 'port', 'tls', 'timeout', 'sftp', 'pysftp_conn_kwargs'
])

_ConnectionConfig = namedtuple('_ConnectionConfig', [
'tls', 'timeout', 'sftp'
])


class RemoteFileSystem(luigi.target.FileSystem):
def __init__(self, host, username=None, password=None, port=None, tls=False, timeout=60, sftp=False, pysftp_conn_kwargs=None):
self.host = host
self.username = username
self.password = password
self.tls = tls
self.timeout = timeout
self.sftp = sftp
self.pysftp_conn_kwargs = pysftp_conn_kwargs or {}

if port is None:
if self.sftp:
self.port = 22
else:
self.port = 21
else:
self.port = port
port = 22 if sftp else 21
self._config = _RemoteFileSystemConfig(
host=host, username=username, password=password, port=port,
tls=tls, timeout=timeout, sftp=sftp,
pysftp_conn_kwargs=pysftp_conn_kwargs or {}
)

def _connect(self):
"""
Log in to ftp.
"""
if self.sftp:
if self._config.sftp:
self._sftp_connect()
else:
self._ftp_connect()
Expand All @@ -75,23 +77,23 @@ def _sftp_connect(self):
except ImportError:
logger.warning("Please install pysftp to use SFTP.")

self.conn = pysftp.Connection(self.host, username=self.username, password=self.password, port=self.port, **self.pysftp_conn_kwargs)
self.conn = pysftp.Connection(self._config.host, username=self._config.username, password=self._config.password, port=self._config.port, **self._config.pysftp_conn_kwargs)

def _ftp_connect(self):
if self.tls:
if self._config.tls:
self.conn = ftplib.FTP_TLS()
else:
self.conn = ftplib.FTP()
self.conn.connect(self.host, self.port, timeout=self.timeout)
self.conn.login(self.username, self.password)
if self.tls:
self.conn.connect(self._config.host, self._config.port, timeout=self._config.timeout)
self.conn.login(self._config.username, self._config.password)
if self._config.tls:
self.conn.prot_p()

def _close(self):
"""
Close ftp connection.
"""
if self.sftp:
if self._config.sftp:
self._sftp_close()
else:
self._ftp_close()
Expand All @@ -112,7 +114,7 @@ def exists(self, path, mtime=None):
"""
self._connect()

if self.sftp:
if self._config.sftp:
exists = self._sftp_exists(path, mtime)
else:
exists = self._ftp_exists(path, mtime)
Expand Down Expand Up @@ -156,7 +158,7 @@ def remove(self, path, recursive=True):
"""
self._connect()

if self.sftp:
if self._config.sftp:
self._sftp_remove(path, recursive)
else:
self._ftp_remove(path, recursive)
Expand Down Expand Up @@ -237,7 +239,7 @@ def put(self, local_path, path, atomic=True):
"""
self._connect()

if self.sftp:
if self._config.sftp:
self._sftp_put(local_path, path, atomic)
else:
self._ftp_put(local_path, path, atomic)
Expand Down Expand Up @@ -298,7 +300,7 @@ def get(self, path, local_path):
# download file
self._connect()

if self.sftp:
if self._config.sftp:
self._sftp_get(path, tmp_local_path)
else:
self._ftp_get(path, tmp_local_path)
Expand All @@ -319,7 +321,7 @@ def listdir(self, path="."):
"""
self._connect()

if self.sftp:
if self._config.sftp:
contents = self._sftp_listdir(path)
else:
contents = self._ftp_listdir(path)
Expand Down Expand Up @@ -377,10 +379,8 @@ def __init__(
self.path = path
self.mtime = mtime
self.format = format
self.tls = tls
self.timeout = timeout
self.sftp = sftp
self._fs = RemoteFileSystem(host, username, password, port, tls, timeout, sftp, pysftp_conn_kwargs)
self._conn_config = _ConnectionConfig(tls, timeout, sftp)
self._fs = RemoteFileSystem(host, username, password, port, self._conn_config.tls, self._conn_config.timeout, self._conn_config.sftp, pysftp_conn_kwargs)

@property
def fs(self):
Expand Down
Loading
Loading