128 lines
5.2 KiB
Python
128 lines
5.2 KiB
Python
"""
|
|
SQLAlchemy 2.x ORM Models for eamco_address_checker.
|
|
|
|
This module defines the database models using SQLAlchemy's DeclarativeBase.
|
|
|
|
Models:
|
|
CustomerCustomer: Customer records with address fields for geocoding
|
|
StreetReference: Known streets by town/state for fuzzy matching corrections
|
|
"""
|
|
|
|
from sqlalchemy import Column, Integer, String, VARCHAR, TIMESTAMP, BOOLEAN, Index
|
|
from sqlalchemy.orm import DeclarativeBase
|
|
|
|
|
|
class Base(DeclarativeBase):
|
|
"""Base class for all SQLAlchemy models."""
|
|
pass
|
|
|
|
|
|
class CustomerCustomer(Base):
|
|
"""
|
|
Customer model representing address and contact information.
|
|
|
|
The verified_at timestamp tracks when the address was last geocoded.
|
|
The correct_address boolean indicates if geocoding was successful.
|
|
|
|
Attributes:
|
|
id: Primary key
|
|
auth_net_profile_id: Authorize.net customer profile ID
|
|
account_number: Customer account number (max 25 chars)
|
|
customer_last_name: Customer's last name (max 250 chars)
|
|
customer_first_name: Customer's first name (max 250 chars)
|
|
customer_town: City/town name (max 140 chars)
|
|
customer_state: Integer mapping to US state abbreviation
|
|
customer_zip: ZIP code (max 25 chars)
|
|
customer_first_call: Timestamp of first customer contact
|
|
customer_email: Customer email address (max 500 chars)
|
|
customer_automatic: Automatic billing flag
|
|
customer_phone_number: Phone number (max 25 chars)
|
|
customer_home_type: Type of residence
|
|
customer_apt: Apartment/unit number (max 140 chars)
|
|
customer_address: Street address (max 1000 chars)
|
|
company_id: Associated company ID
|
|
customer_latitude: Geocoded latitude as string (max 250 chars)
|
|
customer_longitude: Geocoded longitude as string (max 250 chars)
|
|
correct_address: Flag indicating successful geocoding
|
|
verified_at: Timestamp of last verification attempt
|
|
"""
|
|
__tablename__ = "customer_customer"
|
|
__table_args__ = {"schema": "public"}
|
|
|
|
id = Column(Integer, primary_key=True, autoincrement=True)
|
|
auth_net_profile_id = Column(String, unique=True, index=True, nullable=True)
|
|
account_number = Column(VARCHAR(25))
|
|
customer_last_name = Column(VARCHAR(250))
|
|
customer_first_name = Column(VARCHAR(250))
|
|
customer_town = Column(VARCHAR(140))
|
|
customer_state = Column(Integer) # Integer -> 2-letter US state abbreviation
|
|
customer_zip = Column(VARCHAR(25))
|
|
customer_first_call = Column(TIMESTAMP)
|
|
customer_email = Column(VARCHAR(500))
|
|
customer_automatic = Column(Integer)
|
|
customer_phone_number = Column(VARCHAR(25))
|
|
customer_home_type = Column(Integer)
|
|
customer_apt = Column(VARCHAR(140))
|
|
customer_address = Column(VARCHAR(1000))
|
|
company_id = Column(Integer)
|
|
customer_latitude = Column(VARCHAR(250))
|
|
customer_longitude = Column(VARCHAR(250))
|
|
correct_address = Column(BOOLEAN, default=False, nullable=False)
|
|
verified_at = Column(TIMESTAMP, nullable=True) # NEW: Tracks verification timestamp
|
|
|
|
def __repr__(self) -> str:
|
|
return (
|
|
f"<CustomerCustomer(id={self.id}, "
|
|
f"name='{self.customer_first_name} {self.customer_last_name}', "
|
|
f"address='{self.customer_address}', "
|
|
f"verified={self.correct_address})>"
|
|
)
|
|
|
|
|
|
class StreetReference(Base):
|
|
"""
|
|
Reference table of known streets for fuzzy matching address corrections.
|
|
|
|
Streets are populated per town/state from OpenStreetMap data.
|
|
Used to correct misspellings and wrong street suffixes (rd vs dr, etc.)
|
|
when geocoding fails.
|
|
|
|
Attributes:
|
|
id: Primary key
|
|
street_name: Full street name (e.g., "Main Street")
|
|
street_name_normalized: Lowercase, cleaned for matching
|
|
street_number_low: Lowest known street number (if available)
|
|
street_number_high: Highest known street number (if available)
|
|
town: Town/city name
|
|
town_normalized: Lowercase town name for matching
|
|
state: 2-letter state abbreviation (e.g., "MA")
|
|
zip_codes: Comma-separated ZIP codes this street spans
|
|
osm_id: OpenStreetMap way ID for reference
|
|
created_at: When this record was added
|
|
"""
|
|
__tablename__ = "street_reference"
|
|
__table_args__ = (
|
|
Index("ix_street_ref_town_state", "town_normalized", "state"),
|
|
Index("ix_street_ref_name_town", "street_name_normalized", "town_normalized"),
|
|
{"schema": "public"},
|
|
)
|
|
|
|
id = Column(Integer, primary_key=True, autoincrement=True)
|
|
street_name = Column(VARCHAR(500), nullable=False)
|
|
street_name_normalized = Column(VARCHAR(500), nullable=False, index=True)
|
|
street_number_low = Column(Integer, nullable=True)
|
|
street_number_high = Column(Integer, nullable=True)
|
|
town = Column(VARCHAR(140), nullable=False)
|
|
town_normalized = Column(VARCHAR(140), nullable=False)
|
|
state = Column(VARCHAR(2), nullable=False)
|
|
zip_codes = Column(VARCHAR(100), nullable=True)
|
|
osm_id = Column(String, nullable=True, index=True)
|
|
created_at = Column(TIMESTAMP, nullable=False)
|
|
|
|
def __repr__(self) -> str:
|
|
return (
|
|
f"<StreetReference(id={self.id}, "
|
|
f"street='{self.street_name}', "
|
|
f"town='{self.town}', state='{self.state}')>"
|
|
)
|