A typed Python SDK for the Sonny's Carwash Controls Data API.
sonnys-data-client wraps the Sonny's Carwash Controls REST API with a
resource-based interface (client.transactions.list(),
client.customers.get(id)). Every response is returned as a validated Pydantic
v2 model, list calls auto-paginate transparently, and a built-in rate limiter
with exponential-backoff retry keeps your application within the API's
20-request/15-second window.
- Installation
- Quick Start
- Client Configuration
- Resources
- BackOffice
- Error Handling
- Logging
- Multi-Site Usage
- Requirements
pip install git+https://github.com/christopher-nance/Sonnys-Data-API-Clientfrom sonnys_data_client import SonnysClient
with SonnysClient(api_id="your-api-id", api_key="your-api-key") as client:
transactions = client.transactions.list(startDate="2024-01-01", endDate="2024-01-31")
for txn in transactions:
print(txn.transaction_id, txn.total)All resources auto-paginate by default -- calling .list() returns every record
across all pages.
SonnysClient(
api_id: str,
api_key: str,
site_code: str | None = None,
*,
max_retries: int = 3,
backoffice_username: str | None = None,
backoffice_password: str | None = None,
)| Parameter | Type | Default | Description |
|---|---|---|---|
api_id |
str |
required | Sonny's API ID credential. Also doubles as the BackOffice subdomain (e.g. "washu" -> washu.sonnyscontrols.com) when the BackOffice resource is used |
api_key |
str |
required | Sonny's API key credential |
site_code |
str | None |
None |
Optional site code to scope all Data API requests to a single site (ignored by the BackOffice resource) |
max_retries |
int |
3 |
Maximum retry attempts for 429 rate-limit responses (uses exponential backoff) |
backoffice_username |
str | None |
None |
BackOffice (manager portal) username. Required to use client.backoffice.*. Distinct from the API credentials. |
backoffice_password |
str | None |
None |
BackOffice (manager portal) password. |
Use the with statement to ensure the underlying HTTP session is closed
automatically when you are done:
with SonnysClient(api_id="id", api_key="key") as client:
data = client.customers.list()If you prefer manual lifecycle management, call .close() explicitly:
client = SonnysClient(api_id="id", api_key="key")
try:
data = client.customers.list()
finally:
client.close()Credentials are sent as HTTP headers on every request:
X-Sonnys-API-ID-- fromapi_idX-Sonnys-API-Key-- fromapi_keyX-Sonnys-Site-Code-- fromsite_code(omitted whenNone)
Each resource is backed by a data source:
— wraps the Sonny's Data REST API (
trigonapi.sonnyscontrols.com). Uses yourapi_id/api_key.— scrapes the manager-portal web UI (
{api_id}.sonnyscontrols.com). Requires the separatebackoffice_username/backoffice_password.
All list() methods auto-paginate by default -- every page is fetched
transparently and the complete result set is returned. Common query parameters
include startDate, endDate, site, region, limit, and offset.
Methods: list(**params) | get(id)
customers = client.customers.list(startDate="2024-01-01", endDate="2024-01-31")
for c in customers:
print(c.customer_id, c.first_name, c.last_name, c.is_active)
# Get full detail for a single customer
detail = client.customers.get("12345")
print(detail.email, detail.phone, detail.address.city)Returns: list[CustomerListItem] from list() -- fields include
customer_id, first_name, last_name, phone_number, is_active,
created_date. Customer from get() -- adds email, address, company_name,
loyalty_number, birth_date.
Methods: list(**params)
items = client.items.list()
for item in items:
print(item.sku, item.name, item.department_name, item.price_at_site)Returns: list[Item] -- fields include sku, name, department_name,
price_at_site, cost_per_item, is_prompt_for_price, site_location.
Methods: list(**params) | get(id) | get_clock_entries(employee_id, *, start_date, end_date)
employees = client.employees.list()
for emp in employees:
print(emp.employee_id, emp.first_name, emp.last_name)
# Get clock entries for a date range
entries = client.employees.get_clock_entries(
42, start_date="2024-01-01", end_date="2024-01-07"
)
for entry in entries:
print(entry.clock_in, entry.clock_out, entry.regular_hours, entry.site_code)Returns: list[EmployeeListItem] from list() -- fields include
employee_id, first_name, last_name. Employee from get() -- adds
active, start_date, phone, email. list[ClockEntry] from
get_clock_entries() -- fields include clock_in, clock_out,
regular_hours, overtime_hours, regular_rate, site_code.
Methods: list(**params)
sites = client.sites.list()
for site in sites:
print(site.site_id, site.code, site.name, site.timezone)Returns: list[Site] -- fields include site_id, code, name,
timezone. Sites is non-paginated; all sites are returned in a single
request.
Methods: list(**params)
giftcards = client.giftcards.list(startDate="2024-01-01", endDate="2024-01-31")
for gc in giftcards:
print(gc.giftcard_id, gc.number, gc.value, gc.amount_used, gc.site_code)Returns: list[GiftcardListItem] -- fields include giftcard_id,
number, value, amount_used, site_code, complete_date.
Methods: list(**params) | get(id)
washbooks = client.washbooks.list()
for wb in washbooks:
print(wb.id, wb.name, wb.balance, wb.status)
# Get full detail including tags and vehicles
detail = client.washbooks.get("WB-123")
print(detail.customer.first_name, detail.recurring_info.next_bill_date)
for tag in detail.tags:
print(tag.number, tag.enabled)Returns: list[WashbookListItem] from list() -- fields include id,
name, balance, sign_up_date, cancel_date, status, customer_id.
Washbook from get() -- adds customer, recurring_info, tags,
vehicles.
Methods: list(**params) | get(id) | list_status_changes(**params) | list_modifications(**params) | list_details(**params)
accounts = client.recurring.list(startDate="2024-01-01", endDate="2024-01-31")
for acct in accounts:
print(acct.id, acct.name, acct.status_name, acct.billing_site_code)
# Get status changes for a date range
changes = client.recurring.list_status_changes(
startDate="2024-01-01", endDate="2024-01-31"
)
for change in changes:
print(change.recurring_id, change.old_status, "->", change.new_status)Returns: list[RecurringListItem] from list() -- fields include id,
name, status_name, sign_up_date, billing_site_code, customer_id.
Recurring from get() -- adds plan_name, customer, tags, vehicles,
recurring_statuses, recurring_billings. list[RecurringStatusChange] from
list_status_changes(). list[RecurringModification] from
list_modifications(). list[Recurring] from list_details() (full detail
for every account).
Methods: list(**params) | get(id) | list_by_type(item_type, **params) | list_v2(**params) | load_job(*, poll_interval, timeout, **params)
txns = client.transactions.list(startDate="2024-01-01", endDate="2024-01-31")
for txn in txns:
print(txn.trans_id, txn.date, txn.total)
# Filter by type: wash, prepaid-wash, recurring, washbook,
# giftcard, merchandise, house-account
washes = client.transactions.list_by_type(
"wash", startDate="2024-01-01", endDate="2024-01-31"
)list_v2 returns enriched items with extra fields -- customer_id,
is_recurring_plan_sale, is_recurring_plan_redemption, transaction_status:
v2 = client.transactions.list_v2(startDate="2024-01-01", endDate="2024-01-31")
for txn in v2:
print(txn.trans_id, txn.customer_id, txn.transaction_status)load_job submits a batch job and auto-polls until results are ready.
The API caches job data for 20 minutes and limits the date range to 24 hours:
results = client.transactions.load_job(
startDate="2024-01-01",
endDate="2024-01-01",
poll_interval=2.0, # seconds between poll attempts (default 2.0)
timeout=300.0, # max seconds to wait per job (default 300.0)
)
for item in results:
print(item.id, item.complete_date, item.total, item.transaction_status)Returns: list[TransactionListItem] from list() and list_by_type() --
fields include trans_id, trans_number, total, date.
Transaction from get() -- adds items, tenders, discounts,
customer_name, employee_cashier, location_code.
list[TransactionV2ListItem] from list_v2() -- extends list item with
customer_id, is_recurring_plan_sale, is_recurring_plan_redemption,
transaction_status. list[TransactionJobItem] from load_job() -- full
transaction detail plus v2 enrichment fields.
Methods: total_sales(start, end) | total_washes(start, end) | retail_wash_count(start, end) | new_memberships_sold(start, end, *, exclude_ecomm=False) | conversion_rate(start, end, *, exclude_ecomm=False) | total_labor_cost(start, end) | cost_per_car(start, end) | report(start, end, *, exclude_ecomm=False)
Unlike other resources that wrap REST endpoints directly, client.stats
computes business analytics by fetching raw data and aggregating it locally.
Calculations are designed to align with Rinsed: The Car Wash CRM reporting as closely as possible.
All methods accept a date range as ISO-8601 strings or datetime objects:
# Individual metrics
sales = client.stats.total_sales("2026-01-01", "2026-01-31")
print(f"Revenue: ${sales.total:.2f}")
washes = client.stats.total_washes("2026-01-01", "2026-01-31")
print(f"Total washes: {washes.total}, Member: {washes.member_wash_count}")
rate = client.stats.conversion_rate("2026-01-01", "2026-01-31")
print(f"Conversion: {rate.rate:.1%}")
# In-lane conversion only -- excludes online (E-Comm) sign-ups from the
# new-membership numerator (matched via "E-Comm" in salesDeviceName)
in_lane = client.stats.conversion_rate("2026-01-01", "2026-01-31", exclude_ecomm=True)
print(f"In-lane conversion: {in_lane.rate:.1%}")
# Labor cost breakdown
labor = client.stats.total_labor_cost("2026-01-01", "2026-01-31")
print(f"Labor: ${labor.total_cost:.2f} ({labor.total_hours:.1f}h)")
# Cost per car
cpc = client.stats.cost_per_car("2026-01-01", "2026-01-31")
print(f"Cost per car: ${cpc.cost_per_car:.2f}")
# All KPIs in one call (4 bulk + clock entries + ~N detail calls)
rpt = client.stats.report("2026-01-01", "2026-01-31")
print(f"Revenue: ${rpt.sales.total:.2f}, Washes: {rpt.washes.total}")
print(f"New members: {rpt.new_memberships}, Conversion: {rpt.conversion.rate:.1%}")
print(f"Labor: ${rpt.labor.total_cost:.2f}, Cost/car: ${rpt.cost_per_car.cost_per_car:.2f}")Returns: SalesResult from total_sales() -- fields include total,
count, recurring_plan_sales, retail. WashResult from total_washes()
-- fields include total, retail_wash_count, member_wash_count,
eligible_wash_count, free_wash_count. int from retail_wash_count() and
new_memberships_sold(). ConversionResult from conversion_rate() -- fields
include rate, new_memberships, eligible_washes. LaborCostResult from
total_labor_cost() -- fields include total_cost, regular_cost,
overtime_cost, regular_hours, overtime_hours, total_hours,
entry_count. CostPerCarResult from cost_per_car() -- fields include
cost_per_car, total_labor_cost, total_washes. StatsReport from
report() -- bundles sales, washes, new_memberships, conversion,
labor, cost_per_car, period_start, period_end.
Methods: timeclock(start, end, *, site_id=None)
The client.backoffice resource scrapes the Sonny's BackOffice web UI
(e.g. https://washu.sonnyscontrols.com) to retrieve per-shift timeclock
data far faster than client.stats.total_labor_cost() can. The
/report/employee-timesheets page returns every employee across every
site for a whole month in a single authenticated page load.
from sonnys_data_client import SonnysClient
with SonnysClient(
api_id="washu", # doubles as BackOffice subdomain
api_key="your-api-key",
backoffice_username="your.manager.login",
backoffice_password="your-backoffice-password",
) as client:
result = client.backoffice.timeclock("2026-03-01", "2026-03-31")
print(f"{len(result.employees)} employees, ${result.total_wages:,.2f}")
for emp in result.employees[:5]:
for shift in emp.shifts:
print(
f" {emp.employee_name} {shift.date_in} "
f"{shift.time_in}-{shift.time_out} @ {shift.site_code} "
f"({shift.regular_hours:.2f}h)"
)Returns: BackOfficeTimeclockResult -- fields include
period_start, period_end, employees (list of EmployeeTimesheet),
total_regular_hours, total_regular_wages, total_overtime_hours,
total_overtime_wages, total_wages.
Each EmployeeTimesheet contains employee_name, employee_number,
adp_id, shifts (list of TimesheetShift), plus per-employee rollup
totals. Each TimesheetShift preserves the raw punch-in/out times, the
site code, the pay rate and hours, and flags (was_modified,
was_created_in_back_office, comment) for audit entries.
Credentials: The BackOffice user is a separate manager-portal
account, not the API credentials. Supply both backoffice_username and
backoffice_password at client construction. Calling
client.backoffice.timeclock() without them raises
BackOfficeCredentialsError. Existing code that constructs
SonnysClient(api_id, api_key) without BackOffice credentials is
unaffected.
See the BackOffice guide for detailed examples, full field tables, and performance notes.
All exceptions inherit from SonnysError:
from sonnys_data_client import SonnysClient, SonnysError, AuthError, NotFoundError
with SonnysClient(api_id="id", api_key="key") as client:
try:
customer = client.customers.get("12345")
except AuthError as e:
print("Bad credentials:", e.message)
except NotFoundError as e:
print("Customer not found:", e.message)
except SonnysError as e:
print("API error:", e)Exception hierarchy:
SonnysError-- base for all errorsAPIError-- base for API-specific errorsAPIConnectionError-- connection failureAPITimeoutError-- request timeout
APIStatusError-- HTTP error response (hasstatus_code,body,error_type)AuthError-- 403 ForbiddenRateLimitError-- 429 Too Many RequestsValidationError-- 400 / 422 bad requestNotFoundError-- 404 Not FoundServerError-- 500+ server errors
BackOfficeError-- base for BackOffice scraper errorsBackOfficeCredentialsError-- missingbackoffice_username/backoffice_passwordat client constructionBackOfficeLoginError-- BackOffice web UI rejected the credentials or was unreachableBackOfficeScrapeError-- timesheets page HTML did not match expected structure (likely a UI change)
Rate limiting is handled automatically -- the client retries 429 responses with
exponential backoff (up to max_retries, default 3). A built-in rate limiter
also throttles outgoing requests to stay under the API rate limit.
The client uses Python standard logging under the sonnys_data_client logger.
Enable debug output to see HTTP requests, responses, and timing:
import logging
logging.getLogger("sonnys_data_client").setLevel(logging.DEBUG)
logging.basicConfig()Log levels:
- DEBUG -- request method/path/params, response status/elapsed time, rate limiter waits
- WARNING -- 429 rate limit retries
Instantiate separate clients for each set of credentials:
from sonnys_data_client import SonnysClient
washu = SonnysClient(api_id="washu-id", api_key="washu-key")
icon = SonnysClient(api_id="icon-id", api_key="icon-key")
washu_sites = washu.sites.list()
icon_sites = icon.sites.list()
washu.close()
icon.close()Use site_code to scope a client to a single site:
client = SonnysClient(api_id="id", api_key="key", site_code="JOLIET")