ADP Payroll Scheduler

Problem

Payroll in the restaurant industry is a special kind of headache. Staff schedules are irregular, overtime rules vary by state, and tip reporting adds another layer of complexity. The operators I worked with were managing payroll for dozens of employees across multiple locations, and every pay period meant the same drill: log into ADP, pull the reports, cross-reference with POS tip data, generate the summary PDFs, and email them to the accountant.

The process was reliable in the sense that it always got done, but it ate up management time every single pay cycle. More importantly, it was error-prone. Manual steps mean manual mistakes, and payroll mistakes mean unhappy employees or compliance issues.

The goal was straightforward: automate the entire report generation pipeline so that payroll summaries arrive in the right inbox on the right day without anyone lifting a finger.

Approach

I built a serverless pipeline on AWS Lambda that triggers on a schedule, pulls payroll data from the ADP API, processes it, generates formatted PDF reports, and delivers them via email. Python was the right tool here thanks to its mature PDF manipulation and data processing libraries.

The Lambda function runs on a CloudWatch Events schedule aligned with the pay cycle. When triggered, it authenticates with the ADP API using OAuth2 client credentials, fetches the relevant payroll data for the current period, and runs it through a series of transformation steps that handle tip allocation, overtime calculations, and location-based tax adjustments.

The output is a set of clean PDF reports (one per location, plus a consolidated summary) that get delivered to a configured email list and archived in S3 for record-keeping. The whole execution typically completes in under 60 seconds, well within Lambda's timeout limits.

Architecture

The pipeline follows a linear flow: a scheduled CloudWatch trigger invokes the Lambda function, which authenticates with ADP's API to fetch payroll data. The data passes through the PDF parser and report generator, producing formatted documents that are delivered via email and archived to S3.

Key Technical Details

The ADP API integration uses OAuth2 with client credentials flow. Token management is handled by a lightweight wrapper that caches tokens in Lambda's execution context between invocations within the same warm container, avoiding unnecessary re-authentication:

class AdpClient:
    def __init__(self, client_id: str, client_secret: str):
        self._token = None
        self._token_expiry = 0
        self._client_id = client_id
        self._client_secret = client_secret

    def _ensure_token(self):
        if time.time() < self._token_expiry - 60:
            return
        response = requests.post(
            "https://accounts.adp.com/auth/oauth/v2/token",
            data={"grant_type": "client_credentials"},
            auth=(self._client_id, self._client_secret),
        )
        data = response.json()
        self._token = data["access_token"]
        self._token_expiry = time.time() + data["expires_in"]

PDF generation uses ReportLab for creating structured payroll summaries. Each report follows a consistent template with location header, employee breakdown, hours and wages table, tip summary, and tax withholding section. The template system is simple enough that adding new sections or changing the layout doesn't require touching the data pipeline.

Error handling follows a "fail loudly" strategy. If the ADP API returns an error or the PDF generation fails for any location, the Lambda function sends an alert email to the operations team with the specific failure details rather than silently skipping the problematic data. Every execution logs the full request/response cycle to CloudWatch for debugging.

The test suite uses pytest with fixtures that mock the ADP API responses. This was critical because ADP's sandbox environment doesn't perfectly mirror production data shapes, and I wanted confidence that the parsing logic handled real-world payroll records correctly.

@pytest.fixture
def sample_payroll_response():
    return {
        "workers": [
            {
                "name": {"formatted": "Jane Smith"},
                "location": "LOC-001",
                "earnings": {"regular": 1200.00, "overtime": 180.00},
                "tips": {"reported": 450.00, "allocated": 50.00},
            }
        ]
    }

def test_payroll_processing(sample_payroll_response):
    result = process_payroll(sample_payroll_response)
    assert result.total_earnings == 1880.00
    assert result.locations == ["LOC-001"]

Links

• Code samples and architecture details available on request