from typing import Union, List, Literal, Callable, Dict, TypeVar, ParamSpec
from functools import wraps
from pandas import DataFrame, ExcelWriter, NA, Series
from numpy import NaN
SaveFormatT = Literal['xlsx', 'csv', 'parquet']
F_Spec = ParamSpec('F_Spec')
F_Return = TypeVar('F_Return')
def save_to_file(
filepath: str,
date_columns: List[str] = None
) -> Callable[
[Callable[F_Spec, F_Return]],
Callable[F_Spec, F_Return]
]:
def decorator(
func: Callable[F_Spec, F_Return]
) -> Callable[F_Spec, F_Return]:
@wraps(func)
def wrapper(*args: F_Spec.args, **kwargs: F_Spec.kwargs) -> F_Return:
result = func(*args, **kwargs)
file_format: SaveFormatT = kwargs.get('save_as')
if file_format is None:
return result
def convert_timezones(df: DataFrame) -> DataFrame:
for column in df.select_dtypes(include=[
'datetime64[ns, UTC]',
]):
df[column] = df[column].dt.tz_convert(None)
return df
if isinstance(result, DataFrame):
df, sheets = result, func.__name__
elif isinstance(result, Dict):
df, sheets = tuple(result.values()), tuple(result.keys())
assert len(df) == len(sheets), f"""
The number of sheets and the number of dataframes are different
(Sheetlist: {len(sheets)}, df list: {len(df)}).
The number of sheets and dataframes must match.
"""
else:
raise TypeError('Data with not supported format')
df: DataFrame | Dict[str, DataFrame]
sheets: str | List[str]
try:
if file_format == 'xlsx':
LOG.info('Save dataframe(s) to Excel')
with ExcelWriter(
filepath + '.xlsx',
mode='w',
) as writer:
if isinstance(result, DataFrame):
result_xlsx = convert_timezones(result.copy())
result_xlsx.to_excel(
writer, sheet_name=sheets, index=False)
elif isinstance(result, Dict):
for frame, sheet in zip(df, sheets):
frame_xlsx = convert_timezones(frame.copy())
frame_xlsx.to_excel(
writer, sheet_name=sheet, index=False)
if file_format == 'csv':
LOG.info('Save dataframe(s) to CSV')
if isinstance(result, DataFrame):
result.to_csv(filepath + '.csv', encoding='utf-8')
if isinstance(result, Dict):
for frame, sheet in zip(df, sheets):
frame.to_csv(
f'{filepath}_{sheet}.csv', encoding='utf-8')
if file_format == 'parquet':
LOG.info('Save dataframe(s) to Parquet')
print('Save dataframe(s) to Parquet')
if isinstance(result, DataFrame):
result.to_parquet(
filepath + '.parquet', compression='gzip')
if isinstance(result, Dict):
for frame, sheet in zip(df, sheets):
frame.to_parquet(
f'{filepath}_{sheet}.parquet',
compression='gzip'
)
except PermissionError:
print("""
\033[91mHave no permission for write the file.
Maybe file is open, you can't write the file while it's open.
Close file and try again\033[0m
""")
return result
return wrapper
return decorator
Обязательным условием для этого декоратора является то, что входным параметром является DataFrame или Dict[str, DataFrame]. И хотелось бы при желании просто указать его над любой функцией, не беспокоясь об ошибке. Например:
> raise ValueError(
"Excel does not support datetimes with "
"timezones. Please ensure that datetimes "
"are timezone unaware before writing to Excel."
)
E ValueError: Excel does not support datetimes with timezones. Please ensure that datetimes are timezone unaware before writing to Excel.
Однако в некоторых таблицах все еще возникают проблемы с этим декоратором, есть таблицы, в которых не все значения в столбцах имеют тип datetime[ns, tz], например столбцы просто имеют тип объекта, как можно определить эти даты и преобразовать их в допустимый формат и как это можно сделать без цикла?
У меня есть декоратор: [code]from typing import Union, List, Literal, Callable, Dict, TypeVar, ParamSpec from functools import wraps from pandas import DataFrame, ExcelWriter, NA, Series from numpy import NaN
def convert_timezones(df: DataFrame) -> DataFrame: for column in df.select_dtypes(include=[ 'datetime64[ns, UTC]', ]): df[column] = df[column].dt.tz_convert(None) return df
if isinstance(result, DataFrame): df, sheets = result, func.__name__ elif isinstance(result, Dict): df, sheets = tuple(result.values()), tuple(result.keys()) assert len(df) == len(sheets), f""" The number of sheets and the number of dataframes are different (Sheetlist: {len(sheets)}, df list: {len(df)}). The number of sheets and dataframes must match. """ else: raise TypeError('Data with not supported format') df: DataFrame | Dict[str, DataFrame] sheets: str | List[str] try: if file_format == 'xlsx': LOG.info('Save dataframe(s) to Excel') with ExcelWriter( filepath + '.xlsx', mode='w', ) as writer: if isinstance(result, DataFrame): result_xlsx = convert_timezones(result.copy()) result_xlsx.to_excel( writer, sheet_name=sheets, index=False) elif isinstance(result, Dict): for frame, sheet in zip(df, sheets): frame_xlsx = convert_timezones(frame.copy()) frame_xlsx.to_excel( writer, sheet_name=sheet, index=False)
if file_format == 'csv': LOG.info('Save dataframe(s) to CSV') if isinstance(result, DataFrame): result.to_csv(filepath + '.csv', encoding='utf-8') if isinstance(result, Dict): for frame, sheet in zip(df, sheets): frame.to_csv( f'{filepath}_{sheet}.csv', encoding='utf-8')
if file_format == 'parquet': LOG.info('Save dataframe(s) to Parquet') print('Save dataframe(s) to Parquet') if isinstance(result, DataFrame): result.to_parquet( filepath + '.parquet', compression='gzip') if isinstance(result, Dict): for frame, sheet in zip(df, sheets): frame.to_parquet( f'{filepath}_{sheet}.parquet', compression='gzip' )
except PermissionError: print(""" \033[91mHave no permission for write the file. Maybe file is open, you can't write the file while it's open. Close file and try again\033[0m """) return result return wrapper return decorator [/code] Обязательным условием для этого декоратора является то, что входным параметром является DataFrame или Dict[str, DataFrame]. И хотелось бы при желании просто указать его над любой функцией, не беспокоясь об ошибке. Например: [code]> raise ValueError( "Excel does not support datetimes with " "timezones. Please ensure that datetimes " "are timezone unaware before writing to Excel." ) E ValueError: Excel does not support datetimes with timezones. Please ensure that datetimes are timezone unaware before writing to Excel. [/code] Однако в некоторых таблицах все еще возникают проблемы с этим декоратором, есть таблицы, в которых не все значения в столбцах имеют тип datetime[ns, tz], например столбцы просто имеют тип объекта, [b]как можно определить эти даты и преобразовать их в допустимый формат и как это можно сделать без цикла?[/b]