Skip to content

Commit fe948d8

Browse files
fatimarahmanfatimarahman
committed
Revert "Reformatted with black"
This reverts commit 69f12e5.
1 parent 69f12e5 commit fe948d8

27 files changed

Lines changed: 748 additions & 942 deletions

‎CHANGELOG.md‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,6 @@
11
# Changelog
22
## v1.1.6 6/26/24
3-
- Generalized Avro functions and separated encoding/decoding behavior
3+
- Generalized Avro functions and separated encoding/decoding behavior.
44

55
## v1.1.5 6/6/24
66
- Use executemany instead of execute when appropriate in RedshiftClient.execute_transaction

‎src/nypl_py_utils/classes/avro_client.py‎

Lines changed: 43 additions & 54 deletions
Original file line numberDiff line numberDiff line change
@@ -9,53 +9,46 @@
99
from nypl_py_utils.functions.log_helper import create_log
1010
from requests.exceptions import JSONDecodeError, RequestException
1111

12-
1312
class AvroClient:
1413
"""
1514
Base class for Avro schema interaction. Takes as input the
1615
Platform API endpoint from which to fetch the schema in JSON format.
1716
"""
18-
17+
1918
def __init__(self, platform_schema_url):
20-
self.logger = create_log("avro_encoder")
21-
self.schema = avro.schema.parse(self.get_json_schema(platform_schema_url))
22-
19+
self.logger = create_log('avro_encoder')
20+
self.schema = avro.schema.parse(
21+
self.get_json_schema(platform_schema_url))
22+
2323
def get_json_schema(self, platform_schema_url):
2424
"""
2525
Fetches a JSON response from the input Platform API endpoint and
2626
interprets it as an Avro schema.
2727
"""
28-
self.logger.info("Fetching Avro schema from {}".format(platform_schema_url))
28+
self.logger.info('Fetching Avro schema from {}'.format(
29+
platform_schema_url))
2930
try:
3031
response = requests.get(platform_schema_url)
3132
response.raise_for_status()
3233
except RequestException as e:
3334
self.logger.error(
34-
"Failed to retrieve schema from {url}: {error}".format(
35-
url=platform_schema_url, error=e
36-
)
37-
)
35+
'Failed to retrieve schema from {url}: {error}'.format(
36+
url=platform_schema_url, error=e))
3837
raise AvroClientError(
39-
"Failed to retrieve schema from {url}: {error}".format(
40-
url=platform_schema_url, error=e
41-
)
42-
) from None
38+
'Failed to retrieve schema from {url}: {error}'.format(
39+
url=platform_schema_url, error=e)) from None
4340

4441
try:
4542
json_response = response.json()
46-
return json_response["data"]["schema"]
43+
return json_response['data']['schema']
4744
except (JSONDecodeError, KeyError) as e:
4845
self.logger.error(
49-
"Retrieved schema is malformed: {errorType} {errorMessage}".format(
50-
errorType=type(e), errorMessage=e
51-
)
52-
)
46+
'Retrieved schema is malformed: {errorType} {errorMessage}'
47+
.format(errorType=type(e), errorMessage=e))
5348
raise AvroClientError(
54-
"Retrieved schema is malformed: {errorType} {errorMessage}".format(
55-
errorType=type(e), errorMessage=e
56-
)
57-
) from None
58-
49+
'Retrieved schema is malformed: {errorType} {errorMessage}'
50+
.format(errorType=type(e), errorMessage=e)) from None
51+
5952

6053
class AvroEncoder(AvroClient):
6154
"""
@@ -70,17 +63,18 @@ def encode_record(self, record):
7063
Returns the encoded record as a byte string.
7164
"""
7265
self.logger.debug(
73-
"Encoding record using {schema} schema".format(schema=self.schema.name)
74-
)
66+
'Encoding record using {schema} schema'.format(
67+
schema=self.schema.name))
7568
datum_writer = DatumWriter(self.schema)
7669
with BytesIO() as output_stream:
7770
encoder = BinaryEncoder(output_stream)
7871
try:
7972
datum_writer.write(record, encoder)
8073
return output_stream.getvalue()
8174
except AvroException as e:
82-
self.logger.error("Failed to encode record: {}".format(e))
83-
raise AvroClientError("Failed to encode record: {}".format(e)) from None
75+
self.logger.error('Failed to encode record: {}'.format(e))
76+
raise AvroClientError(
77+
'Failed to encode record: {}'.format(e)) from None
8478

8579
def encode_batch(self, record_list):
8680
"""
@@ -89,10 +83,8 @@ def encode_batch(self, record_list):
8983
Returns a list of byte strings where each string is an encoded record.
9084
"""
9185
self.logger.info(
92-
"Encoding ({num_rec}) records using {schema} schema".format(
93-
num_rec=len(record_list), schema=self.schema.name
94-
)
95-
)
86+
'Encoding ({num_rec}) records using {schema} schema'.format(
87+
num_rec=len(record_list), schema=self.schema.name))
9688
encoded_records = []
9789
datum_writer = DatumWriter(self.schema)
9890
with BytesIO() as output_stream:
@@ -104,10 +96,9 @@ def encode_batch(self, record_list):
10496
output_stream.seek(0)
10597
output_stream.truncate(0)
10698
except AvroException as e:
107-
self.logger.error("Failed to encode record: {}".format(e))
99+
self.logger.error('Failed to encode record: {}'.format(e))
108100
raise AvroClientError(
109-
"Failed to encode record: {}".format(e)
110-
) from None
101+
'Failed to encode record: {}'.format(e)) from None
111102
return encoded_records
112103

113104

@@ -119,37 +110,34 @@ class AvroDecoder(AvroClient):
119110

120111
def decode_record(self, record, encoding="binary"):
121112
"""
122-
Decodes a single record represented either as a byte or
113+
Decodes a single record represented either as a byte or
123114
base64 string, using the given Avro schema.
124115
125116
Returns a dictionary where each key is a field in the schema.
126117
"""
127-
self.logger.info(
128-
"Decoding {rec} of type {type} using {schema} schema".format(
129-
rec=record, type=encoding, schema=self.schema.name
130-
)
131-
)
132-
118+
self.logger.info('Decoding {rec} of type {type} using {schema} schema'.format(
119+
rec=record, type=encoding, schema=self.schema.name))
120+
133121
if encoding == "base64":
134122
return self._decode_base64(record)
135123
elif encoding == "binary":
136124
return self._decode_binary(record)
137125
else:
138-
self.logger.error(
139-
"Failed to decode record due to encoding type: {}".format(encoding)
140-
)
141-
raise AvroClientError("Invalid encoding type: {}".format(encoding))
142-
126+
self.logger.error('Failed to decode record due to encoding type: {}'.format(encoding))
127+
raise AvroClientError(
128+
'Invalid encoding type: {}'.format(encoding))
129+
143130
def _decode_base64(self, record):
144-
decoded_data = base64.b64decode(record)
131+
decoded_data = base64.b64decode(record).decode("utf-8")
145132
try:
146133
return json.loads(decoded_data)
147134
except Exception as e:
148135
if isinstance(decoded_data, bytes):
149-
return self._decode_binary(decoded_data)
136+
self._decode_binary(decoded_data)
150137
else:
151-
self.logger.error("Failed to decode record: {}".format(e))
152-
raise AvroClientError("Failed to decode record: {}".format(e)) from None
138+
self.logger.error('Failed to decode record: {}'.format(e))
139+
raise AvroClientError(
140+
'Failed to decode record: {}'.format(e)) from None
153141

154142
def _decode_binary(self, record):
155143
datum_reader = DatumReader(self.schema)
@@ -158,10 +146,11 @@ def _decode_binary(self, record):
158146
try:
159147
return datum_reader.read(decoder)
160148
except Exception as e:
161-
self.logger.error("Failed to decode record: {}".format(e))
162-
raise AvroClientError("Failed to decode record: {}".format(e)) from None
149+
self.logger.error('Failed to decode record: {}'.format(e))
150+
raise AvroClientError(
151+
'Failed to decode record: {}'.format(e)) from None
163152

164153

165154
class AvroClientError(Exception):
166155
def __init__(self, message=None):
167-
self.message = message
156+
self.message = message

‎src/nypl_py_utils/classes/kinesis_client.py‎

Lines changed: 26 additions & 35 deletions
Original file line numberDiff line numberDiff line change
@@ -17,19 +17,20 @@ class KinesisClient:
1717
"""
1818

1919
def __init__(self, stream_arn, batch_size, max_retries=5):
20-
self.logger = create_log("kinesis_client")
20+
self.logger = create_log('kinesis_client')
2121
self.stream_arn = stream_arn
2222
self.batch_size = batch_size
2323
self.max_retries = max_retries
2424

2525
try:
2626
self.kinesis_client = boto3.client(
27-
"kinesis", region_name=os.environ.get("AWS_REGION", "us-east-1")
28-
)
27+
'kinesis', region_name=os.environ.get('AWS_REGION',
28+
'us-east-1'))
2929
except ClientError as e:
30-
self.logger.error("Could not create Kinesis client: {err}".format(err=e))
30+
self.logger.error(
31+
'Could not create Kinesis client: {err}'.format(err=e))
3132
raise KinesisClientError(
32-
"Could not create Kinesis client: {err}".format(err=e)
33+
'Could not create Kinesis client: {err}'.format(err=e)
3334
) from None
3435

3536
def close(self):
@@ -44,11 +45,10 @@ def send_records(self, records):
4445
"""
4546
records_sent_since_pause = 0
4647
for i in range(0, len(records), self.batch_size):
47-
encoded_batch = records[i : i + self.batch_size]
48-
kinesis_records = [
49-
{"Data": record, "PartitionKey": str(int(time.time() * 1000000000))}
50-
for record in encoded_batch
51-
]
48+
encoded_batch = records[i:i + self.batch_size]
49+
kinesis_records = [{'Data': record, 'PartitionKey':
50+
str(int(time.time() * 1000000000))}
51+
for record in encoded_batch]
5252

5353
if records_sent_since_pause + len(encoded_batch) > 1000:
5454
records_sent_since_pause = 0
@@ -63,41 +63,32 @@ def _send_kinesis_format_records(self, kinesis_records, call_count):
6363
"""
6464
if call_count > self.max_retries:
6565
self.logger.error(
66-
"Failed to send records to Kinesis {} times in a row".format(
67-
call_count - 1
68-
)
69-
)
66+
'Failed to send records to Kinesis {} times in a row'.format(
67+
call_count-1))
7068
raise KinesisClientError(
71-
"Failed to send records to Kinesis {} times in a row".format(
72-
call_count - 1
73-
)
74-
) from None
69+
'Failed to send records to Kinesis {} times in a row'.format(
70+
call_count-1)) from None
7571

7672
try:
7773
self.logger.info(
78-
"Sending ({count}) records to {arn} Kinesis stream".format(
79-
count=len(kinesis_records), arn=self.stream_arn
80-
)
81-
)
74+
'Sending ({count}) records to {arn} Kinesis stream'.format(
75+
count=len(kinesis_records), arn=self.stream_arn))
8276
response = self.kinesis_client.put_records(
83-
Records=kinesis_records, StreamARN=self.stream_arn
84-
)
85-
if response["FailedRecordCount"] > 0:
77+
Records=kinesis_records, StreamARN=self.stream_arn)
78+
if response['FailedRecordCount'] > 0:
8679
self.logger.warning(
87-
"Failed to send {} records to Kinesis".format(
88-
response["FailedRecordCount"]
89-
)
90-
)
80+
'Failed to send {} records to Kinesis'.format(
81+
response['FailedRecordCount']))
9182
failed_records = []
92-
for i in range(len(response["Records"])):
93-
if "ErrorCode" in response["Records"][i]:
83+
for i in range(len(response['Records'])):
84+
if 'ErrorCode' in response['Records'][i]:
9485
failed_records.append(kinesis_records[i])
95-
self._send_kinesis_format_records(failed_records, call_count + 1)
86+
self._send_kinesis_format_records(failed_records, call_count+1)
9687
except ClientError as e:
97-
self.logger.error("Error sending records to Kinesis: {}".format(e))
88+
self.logger.error(
89+
'Error sending records to Kinesis: {}'.format(e))
9890
raise KinesisClientError(
99-
"Error sending records to Kinesis: {}".format(e)
100-
) from None
91+
'Error sending records to Kinesis: {}'.format(e)) from None
10192

10293

10394
class KinesisClientError(Exception):

‎src/nypl_py_utils/classes/kms_client.py‎

Lines changed: 11 additions & 15 deletions
Original file line numberDiff line numberDiff line change
@@ -11,17 +11,16 @@ class KmsClient:
1111
"""Client for interacting with a KMS client"""
1212

1313
def __init__(self):
14-
self.logger = create_log("kms_client")
14+
self.logger = create_log('kms_client')
1515

1616
try:
1717
self.kms_client = boto3.client(
18-
"kms", region_name=os.environ.get("AWS_REGION", "us-east-1")
19-
)
18+
'kms', region_name=os.environ.get('AWS_REGION', 'us-east-1'))
2019
except ClientError as e:
21-
self.logger.error("Could not create KMS client: {err}".format(err=e))
20+
self.logger.error(
21+
'Could not create KMS client: {err}'.format(err=e))
2222
raise KmsClientError(
23-
"Could not create KMS client: {err}".format(err=e)
24-
) from None
23+
'Could not create KMS client: {err}'.format(err=e)) from None
2524

2625
def close(self):
2726
self.kms_client.close()
@@ -31,19 +30,16 @@ def decrypt(self, encrypted_text):
3130
This method takes a base 64 KMS-encoded string and uses the KMS client
3231
to decrypt it into a usable string.
3332
"""
34-
self.logger.debug("Decrypting '{}'".format(encrypted_text))
33+
self.logger.debug('Decrypting \'{}\''.format(encrypted_text))
3534
try:
3635
decoded_text = b64decode(encrypted_text)
3736
return self.kms_client.decrypt(CiphertextBlob=decoded_text)[
38-
"Plaintext"
39-
].decode("utf-8")
37+
'Plaintext'].decode('utf-8')
4038
except (ClientError, base64Error, TypeError) as e:
41-
self.logger.error(
42-
"Could not decrypt '{val}': {err}".format(val=encrypted_text, err=e)
43-
)
44-
raise KmsClientError(
45-
"Could not decrypt '{val}': {err}".format(val=encrypted_text, err=e)
46-
) from None
39+
self.logger.error('Could not decrypt \'{val}\': {err}'.format(
40+
val=encrypted_text, err=e))
41+
raise KmsClientError('Could not decrypt \'{val}\': {err}'.format(
42+
val=encrypted_text, err=e)) from None
4743

4844

4945
class KmsClientError(Exception):

0 commit comments

Comments
 (0)