5
\$\begingroup\$

I receive daily files (filename_%m_%d_%Y.csv) from a client and I read those in pandas, process them, and store them in Postgres. Sometimes there are delays and we do not get the data for a few days. The problem is that I have everything on a scheduled task and I do not want to manually check if the files exist or not before I run my code since I am using append mode on Postgres (and do not have constraints set up yet).

My approach here is to read the last few days of data from my existing database, and then find the max date in order to know where to start reading new data from. So if the max date is Jan 22, I will know that the start date for reading my new files is Jan 23 onward. Is this a decent approach?

central_time = dt.datetime.now(pytz.timezone('US/Central'))
start_date = central_time - pd.Timedelta(days = 7)
engine = sa.create_engine('postgresql://usr:pass@localhost:XXXX/db')
df = pd.read_sql(sa.text('SELECT * FROM table WHERE report_date > :date'), engine, params={'date': start_date}, parse_dates = ['report_date'])
latest_date = df.report_date.max() + pd.Timedelta(days = 1)

Up until this week I have been exclusively using .csv files. My approach here was to find the beginning of the month and read all files from the beginning of the month until the current date. This meant that everyday I would overwrite my file (So this month I would read Jan 1 - 27 daily files, and create a January file, then tomorrow I would read Jan 1 - 28 daily files and then overwrite my previous MTD file).

start_date = (central_time - pd.Timedelta(days = 0)) - pd.offsets.MonthBegin()
end_date = central_time
days = ((pd.Timestamp(end_date) - pd.Timestamp(start_date)) / np.timedelta64(1, 'D')).astype(int)

def last_n_days(filename, num_of_days, include_curr = False):
    f = '%m_%d_%Y'
    curr = end_date
    if include_curr:
        yield curr.strftime(f)
    for num in range(num_of_days):
        curr = curr - dt.timedelta(days = 1)
        yield curr.strftime(f)
\$\endgroup\$
  • \$\begingroup\$ Hm, can anyone help me find a way to directly read max(date) in a query with pandas? I am using SQLalchemy with psycopg2 (I am new to this though). I tried a few ways but no luck. \$\endgroup\$ – trench Jan 28 '16 at 17:58
2
\$\begingroup\$

First of all I don't know Python, so I will try my best.

For the first portion of code, why do you put 7 days as offset? If the file delays more than 7 days your script will crash. I would do a SQL query for that. One like this:

Select max(report_date) from table; 

and now you have the latest date.

For the second part your code seems ok, but personally I don't see the reason to update the whole month every day, that will be problematic.

For me, the best solution is find the latest date, I gave you the query up there, and after that advance day by day looking for files up to today.

\$\endgroup\$
  • \$\begingroup\$ I didn't realize I could query directly for the max date. let me try that. If that works it will be perfect. The second part of the code was my method before postgres. I would rewrite the monthly file (my end users wanted the raw data) because if I appended to it there would sometimes be duplicates or missing days, etc. It was safer to just rewrite everything. \$\endgroup\$ – trench Jan 27 '16 at 15:31
  • \$\begingroup\$ Having trouble getting the max date to work in the read_sql query up top. I know you aren't familiar with pandas or python though so I don't think you would be able to direct me. \$\endgroup\$ – trench Jan 28 '16 at 17:59
  • \$\begingroup\$ let me try engine = sa.create_engine('postgresql://usr:pass@localhost:XXXX/db') df = pd.read_sql(sa.text('SELECT max(report_date) as rdate FROM table'), engine) latest_date = df.rdate + pd.Timedelta(days = 1) \$\endgroup\$ – Emiliano Feb 1 '16 at 13:51
  • \$\begingroup\$ notice the alias in sql query and you access like another field \$\endgroup\$ – Emiliano Feb 1 '16 at 13:52

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.