Jake/data_pipeline_for_YNAB
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fixed scheduled transactions

Browse Source
This commit is contained in:
Jake Pullen
2024-08-10 08:29:50 +01:00
parent 696b72988d
commit c9b01e85d5
2 changed files with 5 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pipeline/facts.py
+4 -5
View File
@@ -77,7 +77,9 @@ class FactScheduledTransactions(Facts):
scheduled_transactions_df scheduled_transactions_df
.with_columns([ .with_columns([
pl.col("id").alias("scheduled_transaction_id"), pl.col("id").alias("scheduled_transaction_id"),
pl.col("date").alias("scheduled_transaction_date"), pl.col("date_first").alias("scheduled_transaction_first_date"),
pl.col("date_next").alias("scheduled_transaction_next_date"),
pl.col("frequency").alias("scheduled_transaction_frequency"),
pl.col("amount").alias("scheduled_transaction_amount"), pl.col("amount").alias("scheduled_transaction_amount"),
pl.col("memo").alias("scheduled_transaction_memo"), pl.col("memo").alias("scheduled_transaction_memo"),
pl.col("flag_color").alias("scheduled_transaction_flag_color"), pl.col("flag_color").alias("scheduled_transaction_flag_color"),
@@ -91,13 +93,10 @@ class FactScheduledTransactions(Facts):
(pl.col("amount") / 100).alias("scheduled_transaction_amount"), (pl.col("amount") / 100).alias("scheduled_transaction_amount"),
]) ])
.drop([ .drop([
"transfer_transaction_id", "matched_transaction_id", "import_id",
"subtransactions", "deleted","flag_name","account_name", "subtransactions", "deleted","flag_name","account_name",
"payee_name","category_name","import_payee_name","import_payee_name_original", "payee_name","category_name","ingestion_date"
"debt_transaction_type","ingestion_date"
]) ])
) )
# Write the DataFrame to a new parquet file # Write the DataFrame to a new parquet file
logging.info("Writing the transformed scheduled transactions DataFrame to parquet file") logging.info("Writing the transformed scheduled transactions DataFrame to parquet file")
scheduled_transactions_df.write_parquet(self.config['warehouse_data_path'] + '/scheduled_transactions.parquet') scheduled_transactions_df.write_parquet(self.config['warehouse_data_path'] + '/scheduled_transactions.parquet')
test.py
+1 -1
View File
@@ -1,7 +1,7 @@
import polars as pl import polars as pl
#entities = ['accounts', 'categories', 'months', 'payees', 'transactions', 'scheduled_transactions'] #entities = ['accounts', 'categories', 'months', 'payees', 'transactions', 'scheduled_transactions']
entities = ['transactions'] entities = ['scheduled_transactions']
for entity in entities: for entity in entities: