Мне удалось использовать sqlacodegen для создания моделей SQLAlchemy для существующих таблиц базы данных, что очень помогло.
Есть ли что-нибудь подобное, что я могу сделать с файлами? Я получаю файлы из разных источников и не могу контролировать их содержимое - мне просто говорят, что этот файл необходимо добавить в хранилище данных, а затем отправить в инструмент визуализации, чтобы другие могли создавать отчеты.
- Я использую pandas, чтобы дать мне имена столбцов и угадать тип данных для каждого столбца (это вывод в файл конфигурации yaml)
- Затем я редактирую файл yaml и задаю определенные флаги (convert_integer: true, localize_timestamp: true) для каждого столбца, который затем вызовет соответствующую функцию во время моего шага 'transform'. Например, преобразование данных времени (00:04:45) в секунды, преобразование меток времени в UTC и т. Д. Я также указываю правильный тип данных
- Я преобразую данные и затем пропускаю их через шаблон CINATE TABLE jinja2, который берет имя столбца и тип данных из файла конфигурации yaml.
Моя проблема с этим заключается в том, что я иногда путаюсь и вынужден вносить изменения после создания таблицы. Тогда у меня есть устаревший файл конфигурации yaml, и нет записей о моих изменениях. В некоторых случаях мне также необходимо добавлять столбцы.
Моя цель:
- Чтение CSV-файла с пандами
- Угадайте тип данных для каждого столбца и удалите все недопустимые символы (тире и прочее, которые не работают в имени столбца PostgreSQL)
- Автоматически выводить файл модели SQLAlchemy, который я могу использовать для Alembic для создания исходной таблицы, которая должна быть как можно более сырой.
- Сделайте мои шаги преобразования и затем используйте Alembic для обновления таблицы (переименование столбцов, текст в целое число, метка времени в метку времени с часовым поясом, добавление новых столбцов)
Я думаю, что это может создать таблицу, поэтому я бы сделал это, а затем использовал sqlacodegen для создания файла модели для меня? Или есть способ вывести модель напрямую?
def to_sql(engine, df, table, if_exists):
df[:0].to_sql(table, engine, if_exists=if_exists)