Skip to content

Commit 4bde99f

Browse files
zhongjiajiezhongjiajie
authored
Make airflow/providers pylint compatible (#7802)
1 parent a001489 commit 4bde99f

60 files changed

Lines changed: 427 additions & 304 deletions

File tree

Some content is hidden

Large Commits have some content hidden by default. Use the searchbox below for content that may be hidden.

‎UPDATING.md‎

Lines changed: 19 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -61,6 +61,25 @@ https://developers.google.com/style/inclusive-documentation
6161
6262
-->
6363

64+
### Rename parameter name in PinotAdminHook.create_segment
65+
66+
Rename parameter name from ``format`` to ``segment_format`` in PinotAdminHook function create_segment fro pylint compatible
67+
68+
### Rename parameter name in HiveMetastoreHook.get_partitions
69+
70+
Rename parameter name from ``filter`` to ``partition_filter`` in HiveMetastoreHook function get_partitions for pylint compatible
71+
72+
### Remove unnecessary parameter in FTPHook.list_directory
73+
74+
Remove unnecessary parameter ``nlst`` in FTPHook function list_directory for pylint compatible
75+
76+
### Remove unnecessary parameter in PostgresHook function copy_expert
77+
78+
Remove unnecessary parameter ``open`` in PostgresHook function copy_expert for pylint compatible
79+
80+
### Change parameter name in OpsgenieAlertOperator
81+
82+
Change parameter name from ``visibleTo`` to ``visible_to`` in OpsgenieAlertOperator for pylint compatible
6483

6584
### Use NULL as default value for dag.description
6685

‎airflow/providers/apache/druid/hooks/druid.py‎

Lines changed: 9 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -59,6 +59,9 @@ def __init__(
5959
raise ValueError("Druid timeout should be equal or greater than 1")
6060

6161
def get_conn_url(self):
62+
"""
63+
Get Druid connection url
64+
"""
6265
conn = self.get_connection(self.druid_ingest_conn_id)
6366
host = conn.host
6467
port = conn.port
@@ -82,6 +85,9 @@ def get_auth(self):
8285
return None
8386

8487
def submit_indexing_job(self, json_index_spec: str):
88+
"""
89+
Submit Druid ingestion job
90+
"""
8591
url = self.get_conn_url()
8692

8793
self.log.info("Druid ingestion spec: %s", json_index_spec)
@@ -107,7 +113,7 @@ def submit_indexing_job(self, json_index_spec: str):
107113
# ensure that the job gets killed if the max ingestion time is exceeded
108114
requests.post("{0}/{1}/shutdown".format(url, druid_task_id), auth=self.get_auth())
109115
raise AirflowException('Druid ingestion took more than '
110-
'%s seconds', self.max_ingestion_time)
116+
f'{self.max_ingestion_time} seconds')
111117

112118
time.sleep(self.timeout)
113119

@@ -122,7 +128,7 @@ def submit_indexing_job(self, json_index_spec: str):
122128
raise AirflowException('Druid indexing job failed, '
123129
'check console for more info')
124130
else:
125-
raise AirflowException('Could not get status of the job, got %s', status)
131+
raise AirflowException(f'Could not get status of the job, got {status}')
126132

127133
self.log.info('Successful index')
128134

@@ -138,14 +144,11 @@ class DruidDbApiHook(DbApiHook):
138144
default_conn_name = 'druid_broker_default'
139145
supports_autocommit = False
140146

141-
def __init__(self, *args, **kwargs):
142-
super().__init__(*args, **kwargs)
143-
144147
def get_conn(self):
145148
"""
146149
Establish a connection to druid broker.
147150
"""
148-
conn = self.get_connection(self.druid_broker_conn_id)
151+
conn = self.get_connection(self.druid_broker_conn_id) # pylint: disable=no-member
149152
druid_broker_conn = connect(
150153
host=conn.host,
151154
port=conn.port,

‎airflow/providers/apache/hive/hooks/hive.py‎

Lines changed: 33 additions & 32 deletions
Original file line numberDiff line numberDiff line change
@@ -88,6 +88,7 @@ def __init__(
8888
self.auth = conn.extra_dejson.get('auth', 'noSasl')
8989
self.conn = conn
9090
self.run_as = run_as
91+
self.sub_process = None
9192

9293
if mapred_queue_priority:
9394
mapred_queue_priority = mapred_queue_priority.upper()
@@ -241,24 +242,24 @@ def run_cli(self, hql, schema=None, verbose=True, hive_conf=None):
241242

242243
if verbose:
243244
self.log.info("%s", " ".join(hive_cmd))
244-
sp = subprocess.Popen(
245+
sub_process = subprocess.Popen(
245246
hive_cmd,
246247
stdout=subprocess.PIPE,
247248
stderr=subprocess.STDOUT,
248249
cwd=tmp_dir,
249250
close_fds=True)
250-
self.sp = sp
251+
self.sub_process = sub_process
251252
stdout = ''
252253
while True:
253-
line = sp.stdout.readline()
254+
line = sub_process.stdout.readline()
254255
if not line:
255256
break
256257
stdout += line.decode('UTF-8')
257258
if verbose:
258259
self.log.info(line.decode('UTF-8').strip())
259-
sp.wait()
260+
sub_process.wait()
260261

261-
if sp.returncode:
262+
if sub_process.returncode:
262263
raise AirflowException(stdout)
263264

264265
return stdout
@@ -338,7 +339,7 @@ def load_df(
338339
"""
339340

340341
def _infer_field_types_from_df(df):
341-
DTYPE_KIND_HIVE_TYPE = {
342+
dtype_kind_hive_type = {
342343
'b': 'BOOLEAN', # boolean
343344
'i': 'BIGINT', # signed integer
344345
'u': 'BIGINT', # unsigned integer
@@ -351,10 +352,10 @@ def _infer_field_types_from_df(df):
351352
'V': 'STRING' # void
352353
}
353354

354-
d = OrderedDict()
355+
order_type = OrderedDict()
355356
for col, dtype in df.dtypes.iteritems():
356-
d[col] = DTYPE_KIND_HIVE_TYPE[dtype.kind]
357-
return d
357+
order_type[col] = dtype_kind_hive_type[dtype.kind]
358+
return order_type
358359

359360
if pandas_kwargs is None:
360361
pandas_kwargs = {}
@@ -466,12 +467,15 @@ def load_file(
466467
self.run_cli(hql)
467468

468469
def kill(self):
470+
"""
471+
Kill Hive cli command
472+
"""
469473
if hasattr(self, 'sp'):
470-
if self.sp.poll() is None:
474+
if self.sub_process.poll() is None:
471475
print("Killing the Hive job")
472-
self.sp.terminate()
476+
self.sub_process.terminate()
473477
time.sleep(60)
474-
self.sp.kill()
478+
self.sub_process.kill()
475479

476480

477481
class HiveMetastoreHook(BaseHook):
@@ -488,9 +492,9 @@ def __init__(self, metastore_conn_id='metastore_default'):
488492
def __getstate__(self):
489493
# This is for pickling to work despite the thirft hive client not
490494
# being pickable
491-
d = dict(self.__dict__)
492-
del d['metastore']
493-
return d
495+
state = dict(self.__dict__)
496+
del state['metastore']
497+
return state
494498

495499
def __setstate__(self, d):
496500
self.__dict__.update(d)
@@ -504,18 +508,18 @@ def get_metastore_client(self):
504508
from thrift.transport import TSocket, TTransport
505509
from thrift.protocol import TBinaryProtocol
506510

507-
ms = self._find_valid_server()
511+
conn = self._find_valid_server()
508512

509-
if ms is None:
513+
if not conn:
510514
raise AirflowException("Failed to locate the valid server.")
511515

512-
auth_mechanism = ms.extra_dejson.get('authMechanism', 'NOSASL')
516+
auth_mechanism = conn.extra_dejson.get('authMechanism', 'NOSASL')
513517

514518
if conf.get('core', 'security') == 'kerberos':
515-
auth_mechanism = ms.extra_dejson.get('authMechanism', 'GSSAPI')
516-
kerberos_service_name = ms.extra_dejson.get('kerberos_service_name', 'hive')
519+
auth_mechanism = conn.extra_dejson.get('authMechanism', 'GSSAPI')
520+
kerberos_service_name = conn.extra_dejson.get('kerberos_service_name', 'hive')
517521

518-
conn_socket = TSocket.TSocket(ms.host, ms.port)
522+
conn_socket = TSocket.TSocket(conn.host, conn.port)
519523

520524
if conf.get('core', 'security') == 'kerberos' \
521525
and auth_mechanism == 'GSSAPI':
@@ -526,7 +530,7 @@ def get_metastore_client(self):
526530

527531
def sasl_factory():
528532
sasl_client = sasl.Client()
529-
sasl_client.setAttr("host", ms.host)
533+
sasl_client.setAttr("host", conn.host)
530534
sasl_client.setAttr("service", kerberos_service_name)
531535
sasl_client.init()
532536
return sasl_client
@@ -551,6 +555,7 @@ def _find_valid_server(self):
551555
return conn
552556
else:
553557
self.log.info("Could not connect to %s:%s", conn.host, conn.port)
558+
return None
554559

555560
def get_conn(self):
556561
return self.metastore
@@ -577,10 +582,7 @@ def check_for_partition(self, schema, table, partition):
577582
partitions = client.get_partitions_by_filter(
578583
schema, table, partition, 1)
579584

580-
if partitions:
581-
return True
582-
else:
583-
return False
585+
return bool(partitions)
584586

585587
def check_for_named_partition(self, schema, table, partition_name):
586588
"""
@@ -634,8 +636,7 @@ def get_databases(self, pattern='*'):
634636
with self.metastore as client:
635637
return client.get_databases(pattern)
636638

637-
def get_partitions(
638-
self, schema, table_name, filter=None):
639+
def get_partitions(self, schema, table_name, partition_filter=None):
639640
"""
640641
Returns a list of all partitions in a table. Works only
641642
for tables with less than 32767 (java short max val).
@@ -654,10 +655,10 @@ def get_partitions(
654655
if len(table.partitionKeys) == 0:
655656
raise AirflowException("The table isn't partitioned")
656657
else:
657-
if filter:
658+
if partition_filter:
658659
parts = client.get_partitions_by_filter(
659660
db_name=schema, tbl_name=table_name,
660-
filter=filter, max_parts=HiveMetastoreHook.MAX_PART_COUNT)
661+
filter=partition_filter, max_parts=HiveMetastoreHook.MAX_PART_COUNT)
661662
else:
662663
parts = client.get_partitions(
663664
db_name=schema, tbl_name=table_name,
@@ -770,7 +771,7 @@ def table_exists(self, table_name, db='default'):
770771
try:
771772
self.get_table(table_name, db)
772773
return True
773-
except Exception:
774+
except Exception: # pylint: disable=broad-except
774775
return False
775776

776777

@@ -849,7 +850,7 @@ def _get_results(self, hql, schema='default', fetch_size=None, hive_conf=None):
849850
lowered_statement.startswith('show') or
850851
(lowered_statement.startswith('set') and
851852
'=' not in lowered_statement)):
852-
description = [c for c in cur.description]
853+
description = cur.description
853854
if previous_description and previous_description != description:
854855
message = '''The statements are producing different descriptions:
855856
Current: {}

‎airflow/providers/apache/hive/operators/hive.py‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -66,6 +66,7 @@ class HiveOperator(BaseOperator):
6666
template_ext = ('.hql', '.sql',)
6767
ui_color = '#f0e4ec'
6868

69+
# pylint: disable=too-many-arguments
6970
@apply_defaults
7071
def __init__(
7172
self,
@@ -104,6 +105,9 @@ def __init__(
104105
self.hook = None
105106

106107
def get_hook(self):
108+
"""
109+
Get Hive cli hook
110+
"""
107111
return HiveCliHook(
108112
hive_cli_conn_id=self.hive_cli_conn_id,
109113
run_as=self.run_as,

‎airflow/providers/apache/hive/operators/hive_stats.py‎

Lines changed: 18 additions & 15 deletions
Original file line numberDiff line numberDiff line change
@@ -86,22 +86,25 @@ def __init__(self,
8686
self.dttm = '{{ execution_date.isoformat() }}'
8787

8888
def get_default_exprs(self, col, col_type):
89+
"""
90+
Get default expressions
91+
"""
8992
if col in self.col_blacklist:
9093
return {}
91-
d = {(col, 'non_null'): "COUNT({col})"}
94+
exp = {(col, 'non_null'): f"COUNT({col})"}
9295
if col_type in ['double', 'int', 'bigint', 'float']:
93-
d[(col, 'sum')] = 'SUM({col})'
94-
d[(col, 'min')] = 'MIN({col})'
95-
d[(col, 'max')] = 'MAX({col})'
96-
d[(col, 'avg')] = 'AVG({col})'
96+
exp[(col, 'sum')] = f'SUM({col})'
97+
exp[(col, 'min')] = f'MIN({col})'
98+
exp[(col, 'max')] = f'MAX({col})'
99+
exp[(col, 'avg')] = f'AVG({col})'
97100
elif col_type == 'boolean':
98-
d[(col, 'true')] = 'SUM(CASE WHEN {col} THEN 1 ELSE 0 END)'
99-
d[(col, 'false')] = 'SUM(CASE WHEN NOT {col} THEN 1 ELSE 0 END)'
101+
exp[(col, 'true')] = f'SUM(CASE WHEN {col} THEN 1 ELSE 0 END)'
102+
exp[(col, 'false')] = f'SUM(CASE WHEN NOT {col} THEN 1 ELSE 0 END)'
100103
elif col_type in ['string']:
101-
d[(col, 'len')] = 'SUM(CAST(LENGTH({col}) AS BIGINT))'
102-
d[(col, 'approx_distinct')] = 'APPROX_DISTINCT({col})'
104+
exp[(col, 'len')] = f'SUM(CAST(LENGTH({col}) AS BIGINT))'
105+
exp[(col, 'approx_distinct')] = f'APPROX_DISTINCT({col})'
103106

104-
return {k: v.format(col=col) for k, v in d.items()}
107+
return exp
105108

106109
def execute(self, context=None):
107110
metastore = HiveMetastoreHook(metastore_conn_id=self.metastore_conn_id)
@@ -113,12 +116,12 @@ def execute(self, context=None):
113116
}
114117
for col, col_type in list(field_types.items()):
115118
if self.assignment_func:
116-
d = self.assignment_func(col, col_type)
117-
if d is None:
118-
d = self.get_default_exprs(col, col_type)
119+
assign_exprs = self.assignment_func(col, col_type)
120+
if assign_exprs is None:
121+
assign_exprs = self.get_default_exprs(col, col_type)
119122
else:
120-
d = self.get_default_exprs(col, col_type)
121-
exprs.update(d)
123+
assign_exprs = self.get_default_exprs(col, col_type)
124+
exprs.update(assign_exprs)
122125
exprs.update(self.extra_exprs)
123126
exprs = OrderedDict(exprs)
124127
exprs_str = ",\n ".join([

‎airflow/providers/apache/pig/hooks/pig.py‎

Lines changed: 12 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -40,6 +40,7 @@ def __init__(
4040
conn = self.get_connection(pig_cli_conn_id)
4141
self.pig_properties = conn.extra_dejson.get('pig_properties', '')
4242
self.conn = conn
43+
self.sub_process = None
4344

4445
def run_cli(self, pig, pig_opts=None, verbose=True):
4546
"""
@@ -72,27 +73,30 @@ def run_cli(self, pig, pig_opts=None, verbose=True):
7273

7374
if verbose:
7475
self.log.info("%s", " ".join(pig_cmd))
75-
sp = subprocess.Popen(
76+
sub_process = subprocess.Popen(
7677
pig_cmd,
7778
stdout=subprocess.PIPE,
7879
stderr=subprocess.STDOUT,
7980
cwd=tmp_dir,
8081
close_fds=True)
81-
self.sp = sp
82+
self.sub_process = sub_process
8283
stdout = ''
83-
for line in iter(sp.stdout.readline, b''):
84+
for line in iter(sub_process.stdout.readline, b''):
8485
stdout += line.decode('utf-8')
8586
if verbose:
8687
self.log.info(line.strip())
87-
sp.wait()
88+
sub_process.wait()
8889

89-
if sp.returncode:
90+
if sub_process.returncode:
9091
raise AirflowException(stdout)
9192

9293
return stdout
9394

9495
def kill(self):
95-
if hasattr(self, 'sp'):
96-
if self.sp.poll() is None:
96+
"""
97+
Kill Pig job
98+
"""
99+
if self.sub_process:
100+
if self.sub_process.poll() is None:
97101
print("Killing the Pig job")
98-
self.sp.kill()
102+
self.sub_process.kill()

0 commit comments

Comments
 (0)