Additionally, this style plays way better with version control since the field map isn't just a giant one line string that's hundreds of characters long.
Now you just update the mapping dictionary and can see exactly what changed with a commit.
Edit: Here's a more robust version of the above code that supports arbitrary mappings using lambda expressions (You can also just define normal functions and use those. That's a much better idea):
from collections.abc import Callable, Generator, Sequence
from typing import Any
from arcpy.da import InsertCursor, SearchCursor, Editor
type FieldCalc = Callable[[tuple[Any, ...]], Any]
type FieldNames = Sequence[str]
type FieldMap = dict[str, tuple[FieldCalc, FieldNames] | str]
def process_map(f_map: FieldMap) -> tuple[list[str], list[str]]:
"""Consume a field mapping dictionary and return all fields participating on both sides
Note: Returned values are in order (source, target)
"""
target_fields = list(f_map.keys())
source_fields: list[str] = []
for mapping in f_map.values():
if isinstance(mapping, tuple):
# mapping has a functional component
_, fields = mapping
source_fields.extend(fields)
else:
# mapping is direct
source_fields.append(mapping)
return source_fields, target_fields
# Map the row values to field names
def as_dict(cur: SearchCursor) -> Generator[dict[str, Any]]:
"""Make cursor rows dictionaries"""
for row in cur:
yield dict(zip(cur.fields, row))
def calculate_row(target_fields: Sequence[str], source_row: dict[str, Any], field_map: FieldMap) -> list[Any]:
"""Apply a Row-Wise functional/direct map to a Cursor
Note: Returned list will match the field order supplied by `target_fields`
"""
row_vals: list[Any] = []
for field in target_fields:
mapping = field_map.get(field)
if mapping is None:
raise ValueError(f'Field {field} is not mapped!')
if isinstance(mapping, tuple):
# Functional Map
func, source_fields = mapping
row_vals.append(func(tuple(source_row[f] for f in source_fields)))
else:
# Direct Map
row_vals.append(source_row[mapping])
return row_vals
def main():
# Target Workspace
new_survey = 'SurveysAdded'
sde = r'V:\gislu\_DatabaseConnection\editgis1.lu.user.dc.sde'
survey_plat = 'lu.LU.SURV_SurveyPlat'
field_map: FieldMap = {
'survey_id': (lambda vs: ';'.join(map(str, vs)), ['SURVEY_ID1', 'SURVEY_ID2']),
'survey_typ': (lambda vs: ([v for v in vs if v]+[None]).pop(0), ['SURVEY_TYP', 'ALT_TYP']),
'survey_descr': 'SURVEY_DES',
'survey_year': 'SURVEY_YR',
'PLSS_COORD': (lambda vs: vs[0][:6], ['PLSS_COORD']), # Truncate
'link': (lambda vs: str(vs[0]).replace('http', 'https'), ['LINK']),
'parcel_id': 'PARCEL_NO',
}
source_fields, target_fields = process_map(field_map)
new_rows = list(as_dict(SearchCursor(new_survey, source_fields)))
# You can use multiple context managers in one with block
with Editor(sde), InsertCursor(survey_plat, target_fields) as cur:
for row in new_rows:
cur.insertRow(calculate_row(target_fields, row, field_map))
print(f'{len(new_rows)} records appended')
if __name__ == '__main__':
main()
Edit 2:
Here's a more reasonable way to do the functional mappings:
...
def join(vals: tuple[Any, ...]) -> str:
return ';'.join(map(str, vals))
def first(vals: tuple[Any, ...]) -> Any:
return next(filter(bool, vals), None)
def truncate(vals: tuple[str]) -> str:
return vals[0][:6]
def make_https(vals: tuple[str]) -> str:
return vals[0].replace('http', 'https')
field_map: FieldMap = {
'survey_id': (join, ['SURVEY_ID1', 'SURVEY_ID2']),
'survey_typ': (first, ['SURVEY_TYP', 'ALT_TYP']),
'survey_descr': 'SURVEY_DES',
'survey_year': 'SURVEY_YR',
'PLSS_COORD': (truncate, ['PLSS_COORD']), # Truncate
'link': (make_https, ['LINK']),
'parcel_id': 'PARCEL_NO',
}
...
