Skip to content
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

RM-49 Reduce field_to_sqlalchemy_type complexity #224

Merged
merged 5 commits into from
Mar 9, 2023
Merged

RM-49 Reduce field_to_sqlalchemy_type complexity #224

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
24193fa
RM-49 cast func outside of map
ryantimjohn Mar 8, 2023
11175d8
RM-49 fix flake8 linting
ryantimjohn Mar 8, 2023
c5025c2
Merge branch 'master' into RM-49-records_mover-records-schema-field-s…
ryantimjohn Mar 8, 2023
1870a0c
Update mypy_high_water_mark
ryantimjohn Mar 8, 2023
34f3d57
RM-49 add indents for time lambdas
ryantimjohn Mar 9, 2023
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
2 changes: 1 addition & 1 deletion metrics/mypy_high_water_mark
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1 +1 @@
92.3300
92.2800
33 changes: 15 additions & 18 deletions records_mover/records/schema/field/sqlalchemy.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -167,30 +167,27 @@ def field_to_sqlalchemy_type(field: 'RecordsSchemaField',
field_to_func_map = {
'integer': get_integer_type,
'decimal': get_decimal_type,
'boolean': cast(Callable[['RecordsSchemaField', 'DBDriver'], sqlalchemy.types.TypeEngine],
lambda field, driver: sqlalchemy.sql.sqltypes.BOOLEAN()),
'boolean': lambda field, driver: sqlalchemy.sql.sqltypes.BOOLEAN(),
'string': get_string_type,
'date': cast(Callable[['RecordsSchemaField', 'DBDriver'], sqlalchemy.types.TypeEngine],
lambda field, driver: sqlalchemy.sql.sqltypes.DATE()),
'datetime': cast(Callable[['RecordsSchemaField', 'DBDriver'], sqlalchemy.types.TypeEngine],
lambda field, driver: driver.type_for_date_plus_time(has_tz=False)),
'datetimetz': cast(
Callable[['RecordsSchemaField', 'DBDriver'], sqlalchemy.types.TypeEngine],
lambda field, driver: driver.type_for_date_plus_time(has_tz=True)),
'time': cast(Callable[['RecordsSchemaField', 'DBDriver'], sqlalchemy.types.TypeEngine],
lambda field, driver: sqlalchemy.sql.sqltypes.TIME(timezone=False)
if driver.supports_time_type()
else sqlalchemy.sql.sqltypes.VARCHAR(8)),
'timetz': cast(Callable[['RecordsSchemaField', 'DBDriver'], sqlalchemy.types.TypeEngine],
lambda field, driver: sqlalchemy.sql.sqltypes.TIME(timezone=True)
if driver.supports_time_type()
else sqlalchemy.sql.sqltypes.VARCHAR(8))
'date': lambda field, driver: sqlalchemy.sql.sqltypes.DATE(),
'datetime': lambda field, driver: driver.type_for_date_plus_time(has_tz=False),
'datetimetz': lambda field, driver: driver.type_for_date_plus_time(has_tz=True),
'time': lambda field, driver: (
sqlalchemy.sql.sqltypes.TIME(timezone=False)
if driver.supports_time_type()
else sqlalchemy.sql.sqltypes.VARCHAR(8)),
'timetz': lambda field, driver: (
sqlalchemy.sql.sqltypes.TIME(timezone=True)
if driver.supports_time_type()
else sqlalchemy.sql.sqltypes.VARCHAR(8))
}

func = field_to_func_map.get(field.field_type, None)
if not func:
raise NotImplementedError(f"Teach me how to handle records schema type {field.field_type}")
return func(field, driver)
typed_func = cast(Callable[['RecordsSchemaField', 'DBDriver'], sqlalchemy.types.TypeEngine],
func)
return typed_func(field, driver)


def field_to_sqlalchemy_column(field: 'RecordsSchemaField', driver: 'DBDriver') -> Column:
Expand Down