[AIRFLOW-7117] Honor self.schema in sql_to_gcs as schema to upload #8049
Conversation
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
mik-laj
reviewed
Apr 2, 2020
| from tempfile import NamedTemporaryFile | ||
|
|
||
| import unicodecsv as csv | ||
| from six import string_types |
Member
There was a problem hiding this comment.
We don't use six on master branch anymore.
Contributor
Author
There was a problem hiding this comment.
Woops, I will remove that! Good catch.
mik-laj
reviewed
Apr 2, 2020
Comment on lines
262
to
268
| schema = self.schema or [self.field_to_bigquery(field) for field in cursor.description] | ||
|
|
||
| self.log.info('Using schema for %s', self.schema_filename) | ||
| self.log.debug("Current schema: %s", schema) | ||
| tmp_schema_file_handle = NamedTemporaryFile(delete=True) | ||
| tmp_schema_file_handle.write(json.dumps(schema, sort_keys=True).encode('utf-8')) | ||
| schema_json = json.dumps(schema, sort_keys=True) if isinstance(schema, list) else schema | ||
| tmp_schema_file_handle.write(schema_json.encode('utf-8')) |
Member
There was a problem hiding this comment.
It looks overly complicated. What do you think about the following code?
if self.shcema:
schema_json = self.schema
self.log.info("Using user schema")
else:
self.log.info("Starts generating schema")
schema = self.schema or [self.field_to_bigquery(field) for field in cursor.description]
tmp_schema_file_handle = NamedTemporaryFile(delete=True)
schema_json = json.dumps(schema, sort_keys=True)
self.log.info('Using schema for %s', self.schema_filename)
self.log.debug("Current schema: %s", schema)
tmp_schema_file_handle.write(schema_json.encode('utf-8'))
Contributor
Author
There was a problem hiding this comment.
Thanks for the review! I like the structure.
Just to be clear, self.schema could be either str or list, and we don't want to call json.dumps() on a string. How about this edited version? @mik-laj
if self.shcema:
self.log.info("Using user schema")
schema = self.schema
else:
self.log.info("Starts generating schema")
schema = [self.field_to_bigquery(field) for field in cursor.description]
if isinstance(schema, list):
schema = json.dumps(schema, sort_keys=True)
self.log.info('Using schema for %s', self.schema_filename)
self.log.debug("Current schema: %s", schema)
tmp_schema_file_handle = NamedTemporaryFile(delete=True)
tmp_schema_file_handle.write(schema.encode('utf-8'))
Codecov Report
@@ Coverage Diff @@
## master #8049 +/- ##
==========================================
- Coverage 87.25% 87.01% -0.24%
==========================================
Files 935 937 +2
Lines 45384 45555 +171
==========================================
+ Hits 39601 39641 +40
- Misses 5783 5914 +131
Continue to review full report at Codecov.
|
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
3 participants
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
Issue link: AIRFLOW-7117
Make sure to mark the boxes below before creating PR: [x]
In case of fundamental code change, Airflow Improvement Proposal (AIP) is needed.
In case of a new dependency, check compliance with the ASF 3rd Party License Policy.
In case of backwards incompatible changes please leave a note in UPDATING.md.
Read the Pull Request Guidelines for more information.