diff --git a/airflow/task/task_runner/base_task_runner.py b/airflow/task/task_runner/base_task_runner.py
index 83b6064dad7c8..d04329906026c 100644
--- a/airflow/task/task_runner/base_task_runner.py
+++ b/airflow/task/task_runner/base_task_runner.py
@@ -63,9 +63,7 @@ def __init__(self, local_task_job):
             # want to have to specify them in the sudo call - they would show
             # up in `ps` that way! And run commands now, as the other user
             # might not be able to run the cmds to get credentials
-            cfg_path = tmp_configuration_copy(chmod=0o600,
-                                              include_env=True,
-                                              include_cmds=True)
+            cfg_path = tmp_configuration_copy(chmod=0o600)
 
             # Give ownership of file to user; only they can read and write
             subprocess.call(
@@ -85,9 +83,7 @@ def __init__(self, local_task_job):
             # we are running as the same user, and can pass through environment
             # variables then we don't need to include those in the config copy
             # - the runner can read/execute those values as it needs
-            cfg_path = tmp_configuration_copy(chmod=0o600,
-                                              include_env=False,
-                                              include_cmds=False)
+            cfg_path = tmp_configuration_copy(chmod=0o600)
 
         self._cfg_path = cfg_path
         self._command = popen_prepend + self._task_instance.command_as_list(
diff --git a/airflow/utils/cli_action_loggers.py b/airflow/utils/cli_action_loggers.py
index 64ea2b2deb63d..6c0bd99845412 100644
--- a/airflow/utils/cli_action_loggers.py
+++ b/airflow/utils/cli_action_loggers.py
@@ -65,11 +65,11 @@ def on_pre_execution(**kwargs):
     :return: None
     """
     logging.debug("Calling callbacks: %s", __pre_exec_callbacks)
-    for cb in __pre_exec_callbacks:
+    for callback in __pre_exec_callbacks:
         try:
-            cb(**kwargs)
-        except Exception:
-            logging.exception('Failed on pre-execution callback using %s', cb)
+            callback(**kwargs)
+        except Exception:  # pylint: disable=broad-except
+            logging.exception('Failed on pre-execution callback using %s', callback)
 
 
 def on_post_execution(**kwargs):
@@ -83,11 +83,11 @@ def on_post_execution(**kwargs):
     :return: None
     """
     logging.debug("Calling callbacks: %s", __post_exec_callbacks)
-    for cb in __post_exec_callbacks:
+    for callback in __post_exec_callbacks:
         try:
-            cb(**kwargs)
-        except Exception:
-            logging.exception('Failed on post-execution callback using %s', cb)
+            callback(**kwargs)
+        except Exception:  # pylint: disable=broad-except
+            logging.exception('Failed on post-execution callback using %s', callback)
 
 
 def default_action_log(log, **_):
@@ -102,7 +102,7 @@ def default_action_log(log, **_):
     try:
         with create_session() as session:
             session.add(log)
-    except Exception as error:
+    except Exception as error:  # pylint: disable=broad-except
         logging.warning("Failed to log action with %s", error)
 
 
diff --git a/airflow/utils/configuration.py b/airflow/utils/configuration.py
index a0bc80412252a..3ef7d82ea22e9 100644
--- a/airflow/utils/configuration.py
+++ b/airflow/utils/configuration.py
@@ -24,7 +24,7 @@
 from airflow.configuration import conf
 
 
-def tmp_configuration_copy(chmod=0o600, include_env=True, include_cmds=True):
+def tmp_configuration_copy(chmod=0o600):
     """
     Returns a path for a temporary file including a full copy of the configuration
     settings.
diff --git a/airflow/utils/dates.py b/airflow/utils/dates.py
index edaef032a36c7..5628945ff1d22 100644
--- a/airflow/utils/dates.py
+++ b/airflow/utils/dates.py
@@ -20,7 +20,7 @@
 from datetime import datetime, timedelta
 
 from croniter import croniter
-from dateutil.relativedelta import relativedelta  # noqa: F401 for doctest
+from dateutil.relativedelta import relativedelta  # noqa: F401 for doctest  # pylint: disable=unused-import
 
 from airflow.utils import timezone
 
@@ -34,7 +34,7 @@
 }
 
 
-def date_range(start_date, end_date=None, num=None, delta=None):
+def date_range(start_date, end_date=None, num=None, delta=None):  # pylint: disable=too-many-branches
     """
     Get a set of dates as a list based on a start, end and delta, delta
     can be something that can be added to `datetime.datetime`
@@ -73,12 +73,12 @@ def date_range(start_date, end_date=None, num=None, delta=None):
         end_date = timezone.utcnow()
 
     delta_iscron = False
-    tz = start_date.tzinfo
+    time_zone = start_date.tzinfo
 
     if isinstance(delta, str):
         delta_iscron = True
         if timezone.is_localized(start_date):
-            start_date = timezone.make_naive(start_date, tz)
+            start_date = timezone.make_naive(start_date, time_zone)
         cron = croniter(delta, start_date)
     elif isinstance(delta, timedelta):
         delta = abs(delta)
@@ -88,10 +88,10 @@ def date_range(start_date, end_date=None, num=None, delta=None):
     dates = []
     if end_date:
         if timezone.is_naive(start_date) and not timezone.is_naive(end_date):
-            end_date = timezone.make_naive(end_date, tz)
+            end_date = timezone.make_naive(end_date, time_zone)
         while start_date <= end_date:
             if timezone.is_naive(start_date):
-                dates.append(timezone.make_aware(start_date, tz))
+                dates.append(timezone.make_aware(start_date, time_zone))
             else:
                 dates.append(start_date)
 
@@ -102,20 +102,19 @@ def date_range(start_date, end_date=None, num=None, delta=None):
     else:
         for _ in range(abs(num)):
             if timezone.is_naive(start_date):
-                dates.append(timezone.make_aware(start_date, tz))
+                dates.append(timezone.make_aware(start_date, time_zone))
             else:
                 dates.append(start_date)
 
-            if delta_iscron:
-                if num > 0:
-                    start_date = cron.get_next(datetime)
-                else:
-                    start_date = cron.get_prev(datetime)
+            if delta_iscron and num > 0:
+                start_date = cron.get_next(datetime)
+            elif delta_iscron:
+                start_date = cron.get_prev(datetime)
+            elif num > 0:
+                start_date += delta
             else:
-                if num > 0:
-                    start_date += delta
-                else:
-                    start_date -= delta
+                start_date -= delta
+
     return sorted(dates)
 
 
@@ -140,14 +139,14 @@ def round_time(dt, delta, start_date=timezone.make_aware(datetime.min)):
 
     if isinstance(delta, str):
         # It's cron based, so it's easy
-        tz = start_date.tzinfo
-        start_date = timezone.make_naive(start_date, tz)
+        time_zone = start_date.tzinfo
+        start_date = timezone.make_naive(start_date, time_zone)
         cron = croniter(delta, start_date)
         prev = cron.get_prev(datetime)
         if prev == start_date:
-            return timezone.make_aware(start_date, tz)
+            return timezone.make_aware(start_date, time_zone)
         else:
-            return timezone.make_aware(prev, tz)
+            return timezone.make_aware(prev, time_zone)
 
     # Ignore the microseconds of dt
     dt -= timedelta(microseconds=dt.microsecond)
diff --git a/airflow/utils/db.py b/airflow/utils/db.py
index d1b61cad08b01..464a85109a306 100644
--- a/airflow/utils/db.py
+++ b/airflow/utils/db.py
@@ -21,19 +21,22 @@
 
 from airflow import models, settings
 from airflow.configuration import conf
-from airflow.jobs.base_job import BaseJob  # noqa: F401
+from airflow.jobs.base_job import BaseJob  # noqa: F401  # pylint: disable=unused-import
 from airflow.models import Connection
 from airflow.models.pool import Pool
 # We need to add this model manually to get reset working well
-from airflow.models.serialized_dag import SerializedDagModel  # noqa: F401
+from airflow.models.serialized_dag import SerializedDagModel  # noqa: F401  # pylint: disable=unused-import
 from airflow.utils.log.logging_mixin import LoggingMixin
-from airflow.utils.session import create_session, provide_session  # noqa
+from airflow.utils.session import create_session, provide_session  # noqa  # pylint: disable=unused-import
 
 log = LoggingMixin().log
 
 
 @provide_session
 def merge_conn(conn, session=None):
+    """
+    Add new Connection.
+    """
     if not session.query(Connection).filter(Connection.conn_id == conn.conn_id).first():
         session.add(conn)
         session.commit()
@@ -41,6 +44,9 @@ def merge_conn(conn, session=None):
 
 @provide_session
 def add_default_pool_if_not_exists(session=None):
+    """
+    Add default pool if it does not exist.
+    """
     if not Pool.get_pool(Pool.DEFAULT_POOL_NAME, session=session):
         default_pool = Pool(
             pool=Pool.DEFAULT_POOL_NAME,
@@ -54,6 +60,9 @@ def add_default_pool_if_not_exists(session=None):
 
 @provide_session
 def create_default_connections(session=None):
+    """
+    Create default Airflow connections.
+    """
     merge_conn(
         Connection(
             conn_id="airflow_db", conn_type="mysql", host="mysql", login="root", password="", schema="airflow"
@@ -242,7 +251,7 @@ def create_default_connections(session=None):
             conn_id="segment_default", conn_type="segment", extra='{"write_key": "my-segment-write-key"}'
         ),
         session,
-    ),
+    )
     merge_conn(
         Connection(
             conn_id="azure_data_lake_default",
@@ -278,6 +287,9 @@ def create_default_connections(session=None):
 
 
 def initdb():
+    """
+    Initialize Airflow database.
+    """
     upgradedb()
 
     create_default_connections()
@@ -290,10 +302,13 @@ def initdb():
     models.DAG.deactivate_unknown_dags(dagbag.dags.keys())
 
     from flask_appbuilder.models.sqla import Base
-    Base.metadata.create_all(settings.engine)
+    Base.metadata.create_all(settings.engine)  # pylint: disable=no-member
 
 
 def upgradedb():
+    """
+    Upgrade the database.
+    """
     # alembic adds significant import time, so we import it lazily
     from alembic import command
     from alembic.config import Config
@@ -322,11 +337,12 @@ def resetdb():
 
     connection = settings.engine.connect()
     models.base.Base.metadata.drop_all(connection)
-    mc = MigrationContext.configure(connection)
-    if mc._version.exists(connection):
-        mc._version.drop(connection)
+    migartion_ctx = MigrationContext.configure(connection)
+    version = migartion_ctx._version  # pylint: disable=protected-access
+    if version.exists(connection):
+        version.drop(connection)
 
     from flask_appbuilder.models.sqla import Base
-    Base.metadata.drop_all(connection)
+    Base.metadata.drop_all(connection)  # pylint: disable=no-member
 
     initdb()
diff --git a/airflow/utils/decorators.py b/airflow/utils/decorators.py
index 6016c1236aa0b..17f1546f04105 100644
--- a/airflow/utils/decorators.py
+++ b/airflow/utils/decorators.py
@@ -64,9 +64,7 @@ def wrapper(*args, **kwargs):
             dag_args = copy(dag.default_args) or {}
             dag_params = copy(dag.params) or {}
 
-        params = {}
-        if 'params' in kwargs:
-            params = kwargs['params']
+        params = kwargs.get('params', {})
         dag_params.update(params)
 
         default_args = {}
diff --git a/airflow/utils/email.py b/airflow/utils/email.py
index fa2dc7a1c85a6..74b1e3d33035c 100644
--- a/airflow/utils/email.py
+++ b/airflow/utils/email.py
@@ -92,37 +92,43 @@ def send_email_smtp(to, subject, html_content, files=None,
             part['Content-ID'] = '<%s>' % basename
             msg.attach(part)
 
-    send_MIME_email(smtp_mail_from, recipients, msg, dryrun)
+    send_mime_email(smtp_mail_from, recipients, msg, dryrun)
 
 
-def send_MIME_email(e_from, e_to, mime_msg, dryrun=False):
+def send_mime_email(e_from, e_to, mime_msg, dryrun=False):
+    """
+    Send MIME email.
+    """
     log = LoggingMixin().log
 
-    SMTP_HOST = conf.get('smtp', 'SMTP_HOST')
-    SMTP_PORT = conf.getint('smtp', 'SMTP_PORT')
-    SMTP_STARTTLS = conf.getboolean('smtp', 'SMTP_STARTTLS')
-    SMTP_SSL = conf.getboolean('smtp', 'SMTP_SSL')
-    SMTP_USER = None
-    SMTP_PASSWORD = None
+    smtp_host = conf.get('smtp', 'SMTP_HOST')
+    smtp_port = conf.getint('smtp', 'SMTP_PORT')
+    smtp_starttls = conf.getboolean('smtp', 'SMTP_STARTTLS')
+    smtp_ssl = conf.getboolean('smtp', 'SMTP_SSL')
+    smtp_user = None
+    smtp_password = None
 
     try:
-        SMTP_USER = conf.get('smtp', 'SMTP_USER')
-        SMTP_PASSWORD = conf.get('smtp', 'SMTP_PASSWORD')
+        smtp_user = conf.get('smtp', 'SMTP_USER')
+        smtp_password = conf.get('smtp', 'SMTP_PASSWORD')
     except AirflowConfigException:
         log.debug("No user/password found for SMTP, so logging in with no authentication.")
 
     if not dryrun:
-        s = smtplib.SMTP_SSL(SMTP_HOST, SMTP_PORT) if SMTP_SSL else smtplib.SMTP(SMTP_HOST, SMTP_PORT)
-        if SMTP_STARTTLS:
-            s.starttls()
-        if SMTP_USER and SMTP_PASSWORD:
-            s.login(SMTP_USER, SMTP_PASSWORD)
+        conn = smtplib.SMTP_SSL(smtp_host, smtp_port) if smtp_ssl else smtplib.SMTP(smtp_host, smtp_port)
+        if smtp_starttls:
+            conn.starttls()
+        if smtp_user and smtp_password:
+            conn.login(smtp_user, smtp_password)
         log.info("Sent an alert email to %s", e_to)
-        s.sendmail(e_from, e_to, mime_msg.as_string())
-        s.quit()
+        conn.sendmail(e_from, e_to, mime_msg.as_string())
+        conn.quit()
 
 
 def get_email_address_list(addresses: Union[str, Iterable[str]]) -> List[str]:
+    """
+    Get list of email addresses.
+    """
     if isinstance(addresses, str):
         return _get_email_list_from_str(addresses)
 
diff --git a/airflow/utils/file.py b/airflow/utils/file.py
index 8012792117cd2..4a40124a7245f 100644
--- a/airflow/utils/file.py
+++ b/airflow/utils/file.py
@@ -25,7 +25,10 @@
 from airflow import LoggingMixin, conf
 
 
-def TemporaryDirectory(*args, **kwargs):
+def TemporaryDirectory(*args, **kwargs):  # pylint: disable=invalid-name
+    """
+    This function is deprecated. Please use `tempfile.TemporaryDirectory`
+    """
     import warnings
     from tempfile import TemporaryDirectory as TmpDir
     warnings.warn(
diff --git a/airflow/utils/helpers.py b/airflow/utils/helpers.py
index 8c795a20343ac..be202a8e08976 100644
--- a/airflow/utils/helpers.py
+++ b/airflow/utils/helpers.py
@@ -41,6 +41,9 @@
 
 
 def validate_key(k, max_length=250):
+    """
+    Validates value used as a key.
+    """
     if not isinstance(k, str):
         raise TypeError("The key has to be a string")
     elif len(k) > max_length:
@@ -60,18 +63,21 @@ def alchemy_to_dict(obj):
     """
     if not obj:
         return None
-    d = {}
-    for c in obj.__table__.columns:
-        value = getattr(obj, c.name)
-        if type(value) == datetime:
+    output = {}
+    for col in obj.__table__.columns:
+        value = getattr(obj, col.name)
+        if isinstance(value, datetime):
             value = value.isoformat()
-        d[c.name] = value
-    return d
+        output[col.name] = value
+    return output
 
 
 def ask_yesno(question):
+    """
+    Helper to get yes / no answer from user.
+    """
     yes = {'yes', 'y'}
-    no = {'no', 'n'}
+    no = {'no', 'n'}  # pylint: disable=invalid-name
 
     done = False
     print(question)
@@ -141,7 +147,7 @@ def pprinttable(rows):
     If namedtuple are used, the table will have headers
     """
     if not rows:
-        return
+        return None
     if hasattr(rows[0], '_fields'):  # if namedtuple
         headers = rows[0]._fields
     else:
@@ -164,18 +170,18 @@ def pprinttable(rows):
     pattern = " | ".join(formats)
     hpattern = " | ".join(hformats)
     separator = "-+-".join(['-' * n for n in lens])
-    s = ""
-    s += separator + '\n'
-    s += (hpattern % tuple(headers)) + '\n'
-    s += separator + '\n'
+    tab = ""
+    tab += separator + '\n'
+    tab += (hpattern % tuple(headers)) + '\n'
+    tab += separator + '\n'
 
-    def f(t):
+    def _format(t):
         return "{}".format(t) if isinstance(t, str) else t
 
     for line in rows:
-        s += pattern % tuple(f(t) for t in line) + '\n'
-    s += separator + '\n'
-    return s
+        tab += pattern % tuple(_format(t) for t in line) + '\n'
+    tab += separator + '\n'
+    return tab
 
 
 def reap_process_group(pgid, log, sig=signal.SIGTERM,
@@ -221,10 +227,10 @@ def signal_procs(sig):
     except psutil.NoSuchProcess:
         # The process already exited, but maybe it's children haven't.
         children = []
-        for p in psutil.process_iter():
+        for proc in psutil.process_iter():
             try:
-                if os.getpgid(p.pid) == pgid and p.pid != 0:
-                    children.append(p)
+                if os.getpgid(proc.pid) == pgid and proc.pid != 0:
+                    children.append(proc)
             except OSError:
                 pass
 
@@ -240,8 +246,8 @@ def signal_procs(sig):
     _, alive = psutil.wait_procs(children, timeout=timeout, callback=on_terminate)
 
     if alive:
-        for p in alive:
-            log.warning("process %s did not respond to SIGTERM. Trying SIGKILL", p)
+        for proc in alive:
+            log.warning("process %s did not respond to SIGTERM. Trying SIGKILL", proc)
 
         try:
             signal_procs(signal.SIGKILL)
@@ -251,12 +257,15 @@ def signal_procs(sig):
 
         _, alive = psutil.wait_procs(alive, timeout=timeout, callback=on_terminate)
         if alive:
-            for p in alive:
-                log.error("Process %s (%s) could not be killed. Giving up.", p, p.pid)
+            for proc in alive:
+                log.error("Process %s (%s) could not be killed. Giving up.", proc, proc.pid)
     return returncodes
 
 
 def parse_template_string(template_string):
+    """
+    Parses Jinja template string.
+    """
     if "{{" in template_string:  # jinja mode
         return None, Template(template_string)
     else:
@@ -286,4 +295,7 @@ def render_log_filename(ti, try_number, filename_template):
 
 
 def convert_camel_to_snake(camel_str):
+    """
+    Converts CamelCase to snake_case.
+    """
     return re.sub('(?!^)([A-Z]+)', r'_\1', camel_str).lower()
diff --git a/airflow/utils/json.py b/airflow/utils/json.py
index ebf9ea5b17cc1..070db66e70ff6 100644
--- a/airflow/utils/json.py
+++ b/airflow/utils/json.py
@@ -26,22 +26,30 @@
 
 
 class AirflowJsonEncoder(json.JSONEncoder):
+    """
+    Custom Airflow json encoder implementation.
+    """
+    def __init__(self, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+        self.default = self._default
 
-    def default(self, obj):
-        # convert dates and numpy objects in a json serializable format
+    @staticmethod
+    def _default(obj):
+        """
+        Convert dates and numpy objects in a json serializable format.
+        """
         if isinstance(obj, datetime):
             return obj.strftime('%Y-%m-%dT%H:%M:%SZ')
         elif isinstance(obj, date):
             return obj.strftime('%Y-%m-%d')
-        elif type(obj) in (np.int_, np.intc, np.intp, np.int8, np.int16,
-                           np.int32, np.int64, np.uint8, np.uint16,
-                           np.uint32, np.uint64):
+        elif isinstance(obj, (np.int_, np.intc, np.intp, np.int8, np.int16,
+                              np.int32, np.int64, np.uint8, np.uint16,
+                              np.uint32, np.uint64)):
             return int(obj)
-        elif type(obj) in (np.bool_,):
+        elif isinstance(obj, np.bool_):
             return bool(obj)
-        elif type(obj) in (np.float_, np.float16, np.float32, np.float64,
-                           np.complex_, np.complex64, np.complex128):
+        elif isinstance(obj, (np.float_, np.float16, np.float32, np.float64,
+                              np.complex_, np.complex64, np.complex128)):
             return float(obj)
 
-        # Let the base class default method raise the TypeError
-        return json.JSONEncoder.default(self, obj)
+        raise TypeError(f"Object of type '{obj.__class__.__name__}' is not JSON serializable")
diff --git a/airflow/utils/log/es_task_handler.py b/airflow/utils/log/es_task_handler.py
index 064e1668a7cf1..2dacfb49bf7a4 100644
--- a/airflow/utils/log/es_task_handler.py
+++ b/airflow/utils/log/es_task_handler.py
@@ -57,12 +57,13 @@ def __init__(self, base_log_folder, filename_template,
                  log_id_template, end_of_log_mark,
                  write_stdout, json_format, json_fields,
                  host='localhost:9200',
-                 es_kwargs=conf.getsection("elasticsearch_configs") or {}):
+                 es_kwargs=conf.getsection("elasticsearch_configs")):
         """
         :param base_log_folder: base folder to store logs locally
         :param log_id_template: log id template
         :param host: Elasticsearch host name
         """
+        es_kwargs = es_kwargs or {}
         super().__init__(
             base_log_folder, filename_template)
         self.closed = False
@@ -172,25 +173,28 @@ def es_read(self, log_id, offset, metadata):
         """
 
         # Offset is the unique key for sorting logs given log_id.
-        s = Search(using=self.client) \
+        search = Search(using=self.client) \
             .query('match_phrase', log_id=log_id) \
             .sort('offset')
 
-        s = s.filter('range', offset={'gt': int(offset)})
-        max_log_line = s.count()
+        search = search.filter('range', offset={'gt': int(offset)})
+        max_log_line = search.count()
         if 'download_logs' in metadata and metadata['download_logs'] and 'max_offset' not in metadata:
             try:
-                metadata['max_offset'] = s[max_log_line - 1].execute()[-1].offset if max_log_line > 0 else 0
-            except Exception:
+                if max_log_line > 0:
+                    metadata['max_offset'] = search[max_log_line - 1].execute()[-1].offset
+                else:
+                    metadata['max_offset'] = 0
+            except Exception:  # pylint: disable=broad-except
                 self.log.exception('Could not get current log size with log_id: {}'.format(log_id))
 
         logs = []
         if max_log_line != 0:
             try:
 
-                logs = s[self.MAX_LINE_PER_PAGE * self.PAGE:self.MAX_LINE_PER_PAGE] \
+                logs = search[self.MAX_LINE_PER_PAGE * self.PAGE:self.MAX_LINE_PER_PAGE] \
                     .execute()
-            except Exception as e:
+            except Exception as e:  # pylint: disable=broad-except
                 self.log.exception('Could not read log with log_id: %s, error: %s', log_id, str(e))
 
         return logs
@@ -204,12 +208,15 @@ def set_context(self, ti):
         self.mark_end_on_close = not ti.raw
 
         if self.json_format:
-            self.formatter = JSONFormatter(self.formatter._fmt, json_fields=self.json_fields, extras={
-                'dag_id': str(ti.dag_id),
-                'task_id': str(ti.task_id),
-                'execution_date': self._clean_execution_date(ti.execution_date),
-                'try_number': str(ti.try_number)
-            })
+            self.formatter = JSONFormatter(
+                self.formatter._fmt,  # pylint: disable=protected-access
+                json_fields=self.json_fields,
+                extras={
+                    'dag_id': str(ti.dag_id),
+                    'task_id': str(ti.task_id),
+                    'execution_date': self._clean_execution_date(ti.execution_date),
+                    'try_number': str(ti.try_number)
+                })
 
         if self.write_stdout:
             if self.context_set:
@@ -244,7 +251,7 @@ def close(self):
         # Reopen the file stream, because FileHandler.close() would be called
         # first in logging.shutdown() and the stream in it would be set to None.
         if self.handler.stream is None or self.handler.stream.closed:
-            self.handler.stream = self.handler._open()
+            self.handler.stream = self.handler._open()  # pylint: disable=protected-access
 
         # Mark the end of file using end of log mark,
         # so we know where to stop while auto-tailing.
diff --git a/airflow/utils/log/file_processor_handler.py b/airflow/utils/log/file_processor_handler.py
index b11533c742574..a2b80f308d89d 100644
--- a/airflow/utils/log/file_processor_handler.py
+++ b/airflow/utils/log/file_processor_handler.py
@@ -107,7 +107,7 @@ def _symlink_latest_log_directory(self):
         """
         log_directory = self._get_log_directory()
         latest_log_directory_path = os.path.join(self.base_log_folder, "latest")
-        if os.path.isdir(log_directory):
+        if os.path.isdir(log_directory):  # pylint: disable=too-many-nested-blocks
             try:
                 # if symlink exists but is stale, update it
                 if os.path.islink(latest_log_directory_path):
diff --git a/airflow/utils/log/gcs_task_handler.py b/airflow/utils/log/gcs_task_handler.py
index 7745e7b1f9665..cf14070e2d8f2 100644
--- a/airflow/utils/log/gcs_task_handler.py
+++ b/airflow/utils/log/gcs_task_handler.py
@@ -44,13 +44,16 @@ def __init__(self, base_log_folder, gcs_log_folder, filename_template):
 
     @cached_property
     def hook(self):
+        """
+        Returns GCS hook.
+        """
         remote_conn_id = conf.get('logging', 'REMOTE_LOG_CONN_ID')
         try:
             from airflow.gcp.hooks.gcs import GoogleCloudStorageHook
             return GoogleCloudStorageHook(
                 google_cloud_storage_conn_id=remote_conn_id
             )
-        except Exception as e:
+        except Exception as e:  # pylint: disable=broad-except
             self.log.error(
                 'Could not create a GoogleCloudStorageHook with connection id '
                 '"%s". %s\n\nPlease make sure that airflow[gcp] is installed '
@@ -113,7 +116,7 @@ def _read(self, ti, try_number, metadata=None):
             log = '*** Reading remote log from {}.\n{}\n'.format(
                 remote_loc, remote_log)
             return log, {'end_of_log': True}
-        except Exception as e:
+        except Exception as e:  # pylint: disable=broad-except
             log = '*** Unable to read remote log from {}\n*** {}\n\n'.format(
                 remote_loc, str(e))
             self.log.error(log)
@@ -148,8 +151,8 @@ def gcs_write(self, log, remote_log_location, append=True):
             try:
                 old_log = self.gcs_read(remote_log_location)
                 log = '\n'.join([old_log, log]) if old_log else log
-            except Exception as e:
-                if not hasattr(e, 'resp') or e.resp.get('status') != '404':
+            except Exception as e:  # pylint: disable=broad-except
+                if not hasattr(e, 'resp') or e.resp.get('status') != '404':  # pylint: disable=no-member
                     log = '*** Previous log discarded: {}\n\n'.format(str(e)) + log
 
         try:
@@ -162,7 +165,7 @@ def gcs_write(self, log, remote_log_location, append=True):
                 # closed).
                 tmpfile.flush()
                 self.hook.upload(bkt, blob, tmpfile.name)
-        except Exception as e:
+        except Exception as e:  # pylint: disable=broad-except
             self.log.error('Could not write logs to %s: %s', remote_log_location, e)
 
     @staticmethod
diff --git a/airflow/utils/log/logging_mixin.py b/airflow/utils/log/logging_mixin.py
index daeb93e831988..c090c23ebd002 100644
--- a/airflow/utils/log/logging_mixin.py
+++ b/airflow/utils/log/logging_mixin.py
@@ -42,6 +42,9 @@ def __init__(self, context=None):
 
     @property
     def log(self) -> Logger:
+        """
+        Returns a logger.
+        """
         try:
             # FIXME: LoggingMixin should have a default _log field.
             return self._log  # type: ignore
@@ -133,10 +136,13 @@ def __init__(self, stream):
             self._use_stderr = False
 
         # StreamHandler tries to set self.stream
-        Handler.__init__(self)
+        Handler.__init__(self)  # pylint: disable=non-parent-init-called
 
     @property
     def stream(self):
+        """
+        Returns current stream.
+        """
         if self._use_stderr:
             return sys.stderr
 
diff --git a/airflow/utils/log/s3_task_handler.py b/airflow/utils/log/s3_task_handler.py
index d1b0b0327a550..e060384289b6d 100644
--- a/airflow/utils/log/s3_task_handler.py
+++ b/airflow/utils/log/s3_task_handler.py
@@ -41,11 +41,14 @@ def __init__(self, base_log_folder, s3_log_folder, filename_template):
 
     @cached_property
     def hook(self):
+        """
+        Returns S3Hook.
+        """
         remote_conn_id = conf.get('logging', 'REMOTE_LOG_CONN_ID')
         try:
             from airflow.providers.amazon.aws.hooks.s3 import S3Hook
             return S3Hook(remote_conn_id)
-        except Exception:
+        except Exception:  # pylint: disable=broad-except
             self.log.error(
                 'Could not create an S3Hook with connection id "%s". '
                 'Please make sure that airflow[aws] is installed and '
@@ -122,7 +125,7 @@ def s3_log_exists(self, remote_log_location):
         """
         try:
             return self.hook.get_key(remote_log_location) is not None
-        except Exception:
+        except Exception:  # pylint: disable=broad-except
             pass
         return False
 
@@ -139,7 +142,7 @@ def s3_read(self, remote_log_location, return_error=False):
         """
         try:
             return self.hook.read_key(remote_log_location)
-        except Exception:
+        except Exception:  # pylint: disable=broad-except
             msg = 'Could not read logs from {}'.format(remote_log_location)
             self.log.exception(msg)
             # return error if needed
@@ -170,5 +173,5 @@ def s3_write(self, log, remote_log_location, append=True):
                 replace=True,
                 encrypt=conf.getboolean('logging', 'ENCRYPT_S3_LOGS'),
             )
-        except Exception:
+        except Exception:  # pylint: disable=broad-except
             self.log.exception('Could not write logs to %s', remote_log_location)
diff --git a/airflow/utils/log/wasb_task_handler.py b/airflow/utils/log/wasb_task_handler.py
index a4efa78b0f391..322ec9e27eb7f 100644
--- a/airflow/utils/log/wasb_task_handler.py
+++ b/airflow/utils/log/wasb_task_handler.py
@@ -47,6 +47,9 @@ def __init__(self, base_log_folder, wasb_log_folder, wasb_container,
 
     @cached_property
     def hook(self):
+        """
+        Returns WasbHook.
+        """
         remote_conn_id = conf.get('logging', 'REMOTE_LOG_CONN_ID')
         try:
             from airflow.contrib.hooks.wasb_hook import WasbHook
@@ -130,7 +133,7 @@ def wasb_log_exists(self, remote_log_location):
         """
         try:
             return self.hook.check_for_blob(self.wasb_container, remote_log_location)
-        except Exception:
+        except Exception:  # pylint: disable=broad-except
             pass
         return False
 
diff --git a/airflow/utils/net.py b/airflow/utils/net.py
index 318cfa31b234c..d94fac80c4498 100644
--- a/airflow/utils/net.py
+++ b/airflow/utils/net.py
@@ -24,6 +24,9 @@
 
 
 def get_host_ip_address():
+    """
+    Fetch host ip address.
+    """
     return socket.gethostbyname(socket.getfqdn())
 
 
@@ -46,5 +49,5 @@ def get_hostname():
     # Since we have a callable path, we try to import and run it next.
     module_path, attr_name = callable_path.split(':')
     module = importlib.import_module(module_path)
-    callable = getattr(module, attr_name)
-    return callable()
+    func = getattr(module, attr_name)
+    return func()
diff --git a/airflow/utils/operator_resources.py b/airflow/utils/operator_resources.py
index 08220e1f30578..34fb509d718fb 100644
--- a/airflow/utils/operator_resources.py
+++ b/airflow/utils/operator_resources.py
@@ -59,33 +59,55 @@ def __repr__(self):
 
     @property
     def name(self):
+        """
+        Name of the resource.
+        """
         return self._name
 
     @property
     def units_str(self):
+        """
+        The string representing the units of a resource.
+        """
         return self._units_str
 
     @property
     def qty(self):
+        """
+        The number of units of the specified resource that are required for
+        execution of the operator.
+        """
         return self._qty
 
 
 class CpuResource(Resource):
+    """
+    Represents a CPU requirement in an execution environment for an operator.
+    """
     def __init__(self, qty):
         super().__init__('CPU', 'core(s)', qty)
 
 
 class RamResource(Resource):
+    """
+    Represents a RAM requirement in an execution environment for an operator.
+    """
     def __init__(self, qty):
         super().__init__('RAM', 'MB', qty)
 
 
 class DiskResource(Resource):
+    """
+    Represents a disk requirement in an execution environment for an operator.
+    """
     def __init__(self, qty):
         super().__init__('Disk', 'MB', qty)
 
 
 class GpuResource(Resource):
+    """
+    Represents a GPU requirement in an execution environment for an operator.
+    """
     def __init__(self, qty):
         super().__init__('GPU', 'gpu(s)', qty)
 
diff --git a/airflow/utils/sqlalchemy.py b/airflow/utils/sqlalchemy.py
index bab7f7e075567..870948dd5ee14 100644
--- a/airflow/utils/sqlalchemy.py
+++ b/airflow/utils/sqlalchemy.py
@@ -33,13 +33,16 @@
 
 
 def setup_event_handlers(engine):
+    """
+    Setups event handlers.
+    """
     @event.listens_for(engine, "connect")
-    def connect(dbapi_connection, connection_record):  # pylint: disable=unused-variable
+    def connect(dbapi_connection, connection_record):  # pylint: disable=unused-argument
         connection_record.info['pid'] = os.getpid()
 
     if engine.dialect.name == "sqlite":
         @event.listens_for(engine, "connect")
-        def set_sqlite_pragma(dbapi_connection, connection_record):  # pylint: disable=unused-variable
+        def set_sqlite_pragma(dbapi_connection, connection_record):  # pylint: disable=unused-argument
             cursor = dbapi_connection.cursor()
             cursor.execute("PRAGMA foreign_keys=ON")
             cursor.close()
@@ -47,13 +50,13 @@ def set_sqlite_pragma(dbapi_connection, connection_record):  # pylint: disable=u
     # this ensures sanity in mysql when storing datetimes (not required for postgres)
     if engine.dialect.name == "mysql":
         @event.listens_for(engine, "connect")
-        def set_mysql_timezone(dbapi_connection, connection_record):  # pylint: disable=unused-variable
+        def set_mysql_timezone(dbapi_connection, connection_record):  # pylint: disable=unused-argument
             cursor = dbapi_connection.cursor()
             cursor.execute("SET time_zone = '+00:00'")
             cursor.close()
 
     @event.listens_for(engine, "checkout")
-    def checkout(dbapi_connection, connection_record, connection_proxy):  # pylint: disable=unused-variable
+    def checkout(dbapi_connection, connection_record, connection_proxy):  # pylint: disable=unused-argument
         pid = os.getpid()
         if connection_record.info['pid'] != pid:
             connection_record.connection = connection_proxy.connection = None
@@ -90,6 +93,7 @@ def process_bind_param(self, value, dialect):
                 raise ValueError('naive datetime is disallowed')
 
             return value.astimezone(utc)
+        return None
 
     def process_result_value(self, value, dialect):
         """
@@ -109,7 +113,9 @@ def process_result_value(self, value, dialect):
 
 
 class Interval(TypeDecorator):
-
+    """
+    Base class representing a time interval.
+    """
     impl = Text
 
     attr_keys = {
@@ -121,7 +127,7 @@ class Interval(TypeDecorator):
     }
 
     def process_bind_param(self, value, dialect):
-        if type(value) in self.attr_keys:
+        if isinstance(value, tuple(self.attr_keys)):
             attrs = {
                 key: getattr(value, key)
                 for key in self.attr_keys[type(value)]
diff --git a/airflow/utils/state.py b/airflow/utils/state.py
index 71b5cbcc06b88..083c99d380e42 100644
--- a/airflow/utils/state.py
+++ b/airflow/utils/state.py
@@ -79,10 +79,16 @@ class State:
 
     @classmethod
     def color(cls, state):
+        """
+        Returns color for a state.
+        """
         return cls.state_color.get(state, 'white')
 
     @classmethod
     def color_fg(cls, state):
+        """
+        Black&white colors for a state.
+        """
         color = cls.color(state)
         if color in ['green', 'red']:
             return 'white'
diff --git a/airflow/utils/timeout.py b/airflow/utils/timeout.py
index 6a50ebe67815f..89cabb74a7fcf 100644
--- a/airflow/utils/timeout.py
+++ b/airflow/utils/timeout.py
@@ -24,7 +24,7 @@
 from airflow.utils.log.logging_mixin import LoggingMixin
 
 
-class timeout(LoggingMixin):
+class timeout(LoggingMixin):  # pylint: disable=invalid-name
     """
     To be used in a ``with`` block and timeout its content.
     """
@@ -34,6 +34,9 @@ def __init__(self, seconds=1, error_message='Timeout'):
         self.error_message = error_message + ', PID: ' + str(os.getpid())
 
     def handle_timeout(self, signum, frame):
+        """
+        Logs information and raises AirflowTaskTimeout.
+        """
         self.log.error("Process timed out, PID: %s", str(os.getpid()))
         raise AirflowTaskTimeout(self.error_message)
 
@@ -45,7 +48,7 @@ def __enter__(self):
             self.log.warning("timeout can't be used in the current context")
             self.log.exception(e)
 
-    def __exit__(self, type, value, traceback):
+    def __exit__(self, type_, value, traceback):
         try:
             signal.alarm(0)
         except ValueError as e:
diff --git a/airflow/utils/timezone.py b/airflow/utils/timezone.py
index b2278fe3d6164..302d75f25cac8 100644
--- a/airflow/utils/timezone.py
+++ b/airflow/utils/timezone.py
@@ -59,10 +59,10 @@ def utcnow():
     # pendulum utcnow() is not used as that sets a TimezoneInfo object
     # instead of a Timezone. This is not pickable and also creates issues
     # when using replace()
-    d = dt.datetime.utcnow()
-    d = d.replace(tzinfo=utc)
+    date = dt.datetime.utcnow()
+    date = date.replace(tzinfo=utc)
 
-    return d
+    return date
 
 
 def utc_epoch():
@@ -75,10 +75,10 @@ def utc_epoch():
     # pendulum utcnow() is not used as that sets a TimezoneInfo object
     # instead of a Timezone. This is not pickable and also creates issues
     # when using replace()
-    d = dt.datetime(1970, 1, 1)
-    d = d.replace(tzinfo=utc)
+    date = dt.datetime(1970, 1, 1)
+    date = date.replace(tzinfo=utc)
 
-    return d
+    return date
 
 
 def convert_to_utc(value):
@@ -145,16 +145,16 @@ def make_naive(value, timezone=None):
     if is_naive(value):
         raise ValueError("make_naive() cannot be applied to a naive datetime")
 
-    o = value.astimezone(timezone)
+    date = value.astimezone(timezone)
 
     # cross library compatibility
-    naive = dt.datetime(o.year,
-                        o.month,
-                        o.day,
-                        o.hour,
-                        o.minute,
-                        o.second,
-                        o.microsecond)
+    naive = dt.datetime(date.year,
+                        date.month,
+                        date.day,
+                        date.hour,
+                        date.minute,
+                        date.second,
+                        date.microsecond)
 
     return naive
 
diff --git a/airflow/utils/trigger_rule.py b/airflow/utils/trigger_rule.py
index 576aff665bc99..5b9329081d0cb 100644
--- a/airflow/utils/trigger_rule.py
+++ b/airflow/utils/trigger_rule.py
@@ -21,6 +21,9 @@
 
 
 class TriggerRule:
+    """
+    Class with task's trigger rules.
+    """
     ALL_SUCCESS = 'all_success'
     ALL_FAILED = 'all_failed'
     ALL_DONE = 'all_done'
@@ -30,14 +33,20 @@ class TriggerRule:
     NONE_SKIPPED = 'none_skipped'
     DUMMY = 'dummy'
 
-    _ALL_TRIGGER_RULES = set()  # type: Set[str]
+    _ALL_TRIGGER_RULES: Set[str] = set()
 
     @classmethod
     def is_valid(cls, trigger_rule):
+        """
+        Validates a trigger rule.
+        """
         return trigger_rule in cls.all_triggers()
 
     @classmethod
     def all_triggers(cls):
+        """
+        Returns all trigger rules.
+        """
         if not cls._ALL_TRIGGER_RULES:
             cls._ALL_TRIGGER_RULES = {
                 getattr(cls, attr)
diff --git a/airflow/utils/weight_rule.py b/airflow/utils/weight_rule.py
index 1424781159b73..13bf94236c1cf 100644
--- a/airflow/utils/weight_rule.py
+++ b/airflow/utils/weight_rule.py
@@ -21,6 +21,9 @@
 
 
 class WeightRule:
+    """
+    Weight rules.
+    """
     DOWNSTREAM = 'downstream'
     UPSTREAM = 'upstream'
     ABSOLUTE = 'absolute'
@@ -29,10 +32,16 @@ class WeightRule:
 
     @classmethod
     def is_valid(cls, weight_rule):
+        """
+        Check if weight rule is valid.
+        """
         return weight_rule in cls.all_weight_rules()
 
     @classmethod
     def all_weight_rules(cls):
+        """
+        Returns all weight rules
+        """
         if not cls._ALL_WEIGHT_RULES:
             cls._ALL_WEIGHT_RULES = {
                 getattr(cls, attr)
diff --git a/scripts/ci/pylint_todo.txt b/scripts/ci/pylint_todo.txt
index d6d51aa66ce88..a0875dab0be0d 100644
--- a/scripts/ci/pylint_todo.txt
+++ b/scripts/ci/pylint_todo.txt
@@ -198,34 +198,6 @@
 ./airflow/ti_deps/deps/task_concurrency_dep.py
 ./airflow/ti_deps/deps/trigger_rule_dep.py
 ./airflow/ti_deps/deps/valid_state_dep.py
-./airflow/utils/asciiart.py
-./airflow/utils/cli_action_loggers.py
-./airflow/utils/compression.py
-./airflow/utils/configuration.py
-./airflow/utils/dates.py
-./airflow/utils/db.py
-./airflow/utils/decorators.py
-./airflow/utils/email.py
-./airflow/utils/file.py
-./airflow/utils/helpers.py
-./airflow/utils/json.py
-./airflow/utils/log/es_task_handler.py
-./airflow/utils/log/file_processor_handler.py
-./airflow/utils/log/gcs_task_handler.py
-./airflow/utils/log/logging_mixin.py
-./airflow/utils/log/s3_task_handler.py
-./airflow/utils/log/wasb_task_handler.py
-./airflow/utils/module_loading.py
-./airflow/utils/net.py
-./airflow/utils/operator_helpers.py
-./airflow/utils/operator_resources.py
-./airflow/utils/sqlalchemy.py
-./airflow/utils/state.py
-./airflow/utils/tests.py
-./airflow/utils/timeout.py
-./airflow/utils/timezone.py
-./airflow/utils/trigger_rule.py
-./airflow/utils/weight_rule.py
 ./airflow/version.py
 ./airflow/www/api/experimental/endpoints.py
 ./airflow/www/app.py
diff --git a/tests/test_utils/mock_operators.py b/tests/test_utils/mock_operators.py
index ed6386bd1f55a..b5ae985d0968c 100644
--- a/tests/test_utils/mock_operators.py
+++ b/tests/test_utils/mock_operators.py
@@ -93,9 +93,6 @@ def get_link(self, operator, dttm):
 
 
 class CustomOpLink(BaseOperatorLink):
-    """
-    Operator Link for Apache Airflow Website
-    """
     name = 'Google Custom'
 
     def get_link(self, operator, dttm):
diff --git a/tests/utils/test_email.py b/tests/utils/test_email.py
index 3388e7ba2797c..e6d55cce49706 100644
--- a/tests/utils/test_email.py
+++ b/tests/utils/test_email.py
@@ -101,7 +101,7 @@ class TestEmailSmtp(unittest.TestCase):
     def setUp(self):
         conf.set('smtp', 'SMTP_SSL', 'False')
 
-    @mock.patch('airflow.utils.email.send_MIME_email')
+    @mock.patch('airflow.utils.email.send_mime_email')
     def test_send_smtp(self, mock_send_mime):
         attachment = tempfile.NamedTemporaryFile()
         attachment.write(b'attachment')
@@ -120,7 +120,7 @@ def test_send_smtp(self, mock_send_mime):
         mimeapp = MIMEApplication('attachment')
         self.assertEqual(mimeapp.get_payload(), msg.get_payload()[-1].get_payload())
 
-    @mock.patch('airflow.utils.email.send_MIME_email')
+    @mock.patch('airflow.utils.email.send_mime_email')
     def test_send_smtp_with_multibyte_content(self, mock_send_mime):
         utils.email.send_email_smtp('to', 'subject', '🔥', mime_charset='utf-8')
         self.assertTrue(mock_send_mime.called)
@@ -129,7 +129,7 @@ def test_send_smtp_with_multibyte_content(self, mock_send_mime):
         mimetext = MIMEText('🔥', 'mixed', 'utf-8')
         self.assertEqual(mimetext.get_payload(), msg.get_payload()[0].get_payload())
 
-    @mock.patch('airflow.utils.email.send_MIME_email')
+    @mock.patch('airflow.utils.email.send_mime_email')
     def test_send_bcc_smtp(self, mock_send_mime):
         attachment = tempfile.NamedTemporaryFile()
         attachment.write(b'attachment')
@@ -154,7 +154,7 @@ def test_send_mime(self, mock_smtp, mock_smtp_ssl):
         mock_smtp.return_value = mock.Mock()
         mock_smtp_ssl.return_value = mock.Mock()
         msg = MIMEMultipart()
-        utils.email.send_MIME_email('from', 'to', msg, dryrun=False)
+        utils.email.send_mime_email('from', 'to', msg, dryrun=False)
         mock_smtp.assert_called_once_with(
             conf.get('smtp', 'SMTP_HOST'),
             conf.getint('smtp', 'SMTP_PORT'),
@@ -173,7 +173,7 @@ def test_send_mime_ssl(self, mock_smtp, mock_smtp_ssl):
         mock_smtp.return_value = mock.Mock()
         mock_smtp_ssl.return_value = mock.Mock()
         with conf_vars({('smtp', 'smtp_ssl'): 'True'}):
-            utils.email.send_MIME_email('from', 'to', MIMEMultipart(), dryrun=False)
+            utils.email.send_mime_email('from', 'to', MIMEMultipart(), dryrun=False)
         self.assertFalse(mock_smtp.called)
         mock_smtp_ssl.assert_called_once_with(
             conf.get('smtp', 'SMTP_HOST'),
@@ -189,7 +189,7 @@ def test_send_mime_noauth(self, mock_smtp, mock_smtp_ssl):
             ('smtp', 'smtp_user'): None,
             ('smtp', 'smtp_password'): None,
         }):
-            utils.email.send_MIME_email('from', 'to', MIMEMultipart(), dryrun=False)
+            utils.email.send_mime_email('from', 'to', MIMEMultipart(), dryrun=False)
         self.assertFalse(mock_smtp_ssl.called)
         mock_smtp.assert_called_once_with(
             conf.get('smtp', 'SMTP_HOST'),
@@ -200,6 +200,6 @@ def test_send_mime_noauth(self, mock_smtp, mock_smtp_ssl):
     @mock.patch('smtplib.SMTP_SSL')
     @mock.patch('smtplib.SMTP')
     def test_send_mime_dryrun(self, mock_smtp, mock_smtp_ssl):
-        utils.email.send_MIME_email('from', 'to', MIMEMultipart(), dryrun=True)
+        utils.email.send_mime_email('from', 'to', MIMEMultipart(), dryrun=True)
         self.assertFalse(mock_smtp.called)
         self.assertFalse(mock_smtp_ssl.called)