f8cd37ac8c
Includes: API (Express/TypeScript), Astro site, Python workers, document generators, FCC compliance tools, Canada CRTC formation, Ansible infrastructure, and deployment scripts. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
313 lines
12 KiB
Python
313 lines
12 KiB
Python
"""Unified jurisdiction abstraction for US states + Canadian provinces.
|
||
|
||
This module sits alongside the legacy `scripts.formation.states` registry
|
||
and reads from the `jurisdictions` Postgres table (migration 066). It's
|
||
the canonical source of jurisdiction metadata going forward.
|
||
|
||
Why we have both:
|
||
- `scripts.formation.states` still owns per-jurisdiction Playwright
|
||
adapters (`adapter.py` + `config.py` per state) because those files
|
||
contain the hand-written CSS selectors + portal-specific flows.
|
||
- `scripts.formation.jurisdictions` owns the *data* (currency, country,
|
||
entity types, portal URL, NWRA wholesale pricing) that's
|
||
jurisdiction-agnostic and read from the DB.
|
||
|
||
The two are joined by state code. `JurisdictionConfig.adapter()` returns
|
||
the legacy adapter for a code so callers don't have to care.
|
||
|
||
Usage:
|
||
|
||
from scripts.formation.jurisdictions import get_jurisdiction
|
||
|
||
j = get_jurisdiction("WY")
|
||
j.country # 'US'
|
||
j.currency # 'USD'
|
||
j.entity_types # [{'code':'llc','label':'LLC'}, ...]
|
||
j.foreign_qualification_fee_cents("llc") # from state_filing_fees
|
||
adapter = j.adapter() # legacy StatePortal instance
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import logging
|
||
import os
|
||
from dataclasses import dataclass, field
|
||
from functools import lru_cache
|
||
from typing import Optional
|
||
|
||
import psycopg2
|
||
import psycopg2.extras
|
||
|
||
LOG = logging.getLogger("formation.jurisdictions")
|
||
|
||
|
||
# ────────────────────────────────────────────────────────────────────── #
|
||
# Data classes
|
||
# ────────────────────────────────────────────────────────────────────── #
|
||
|
||
|
||
@dataclass
|
||
class EntityTypeSpec:
|
||
"""One entity type a jurisdiction recognizes."""
|
||
code: str # 'llc' | 'corporation' | 'ltd' | 'inc' | ...
|
||
label: str
|
||
|
||
|
||
@dataclass
|
||
class JurisdictionConfig:
|
||
"""Unified config for a single US state / DC / Canadian province.
|
||
|
||
Fields mirror the `jurisdictions` table. `state_filing_fees` data is
|
||
lazy-loaded via the helper methods so we don't pay a second DB hit
|
||
on every access.
|
||
"""
|
||
code: str
|
||
name: str
|
||
country: str # 'US' | 'CA'
|
||
kind: str # 'state' | 'district' | 'province' | 'territory'
|
||
currency: str # 'USD' | 'CAD'
|
||
timezone: Optional[str] = None
|
||
|
||
portal_name: Optional[str] = None
|
||
portal_url: Optional[str] = None
|
||
portal_login_required: bool = False
|
||
|
||
entity_types: list[EntityTypeSpec] = field(default_factory=list)
|
||
|
||
supports_foreign_qualification: bool = True
|
||
foreign_qual_portal_url: Optional[str] = None
|
||
foreign_qual_requires_coa: bool = True
|
||
|
||
nwra_foreign_qual_wholesale_cents: Optional[int] = None
|
||
|
||
notes: Optional[str] = None
|
||
|
||
# ────────────────────────────────────────────────────────────────── #
|
||
# Fee lookups — read on demand from state_filing_fees
|
||
# ────────────────────────────────────────────────────────────────── #
|
||
|
||
def foreign_qualification_fee_cents(self, entity_type: str) -> Optional[int]:
|
||
"""Return target-state's foreign qualification fee for this entity type."""
|
||
col = _FOREIGN_QUAL_FEE_COL.get(_normalize_entity_type(entity_type))
|
||
if not col:
|
||
return None
|
||
row = _query_one(
|
||
f"SELECT {col} AS fee FROM state_filing_fees WHERE state_code = %s",
|
||
(self.code,),
|
||
)
|
||
return int(row["fee"]) if row and row["fee"] is not None else None
|
||
|
||
def formation_fee_cents(self, entity_type: str) -> Optional[int]:
|
||
"""Return home-state's formation fee for this entity type."""
|
||
col = _FORMATION_FEE_COL.get(_normalize_entity_type(entity_type))
|
||
if not col:
|
||
return None
|
||
row = _query_one(
|
||
f"SELECT {col} AS fee FROM state_filing_fees WHERE state_code = %s",
|
||
(self.code,),
|
||
)
|
||
return int(row["fee"]) if row and row["fee"] is not None else None
|
||
|
||
def expedited_fee_cents(self) -> Optional[int]:
|
||
"""Return expedited fee in cents, normalized from the seeded value.
|
||
|
||
`state_filing_fees.expedited_fee` was seeded inconsistently — for
|
||
some states it was stored as dollars × 10000 (the DB convention
|
||
used by `expedited_fee_cents` in formation_orders is cents), so
|
||
we divide by 100 if the stored value is suspiciously large. See
|
||
the same normalization in `scripts.workers.crypto_offramp.sizer`.
|
||
"""
|
||
row = _query_one(
|
||
"SELECT expedited_fee FROM state_filing_fees WHERE state_code = %s",
|
||
(self.code,),
|
||
)
|
||
if not row or row["expedited_fee"] is None:
|
||
return None
|
||
raw = int(row["expedited_fee"])
|
||
return raw // 100 if raw > 50000 else raw
|
||
|
||
def requires_publication(self) -> bool:
|
||
"""Some states (NY, AZ, NE) require newspaper publication after filing."""
|
||
row = _query_one(
|
||
"SELECT publication_required FROM state_filing_fees WHERE state_code = %s",
|
||
(self.code,),
|
||
)
|
||
return bool(row and row.get("publication_required"))
|
||
|
||
# ────────────────────────────────────────────────────────────────── #
|
||
# Adapter bridge — return the legacy StatePortal for this code.
|
||
# ────────────────────────────────────────────────────────────────── #
|
||
|
||
def adapter(self):
|
||
"""Dynamically import the state's StatePortal adapter.
|
||
|
||
Bridges to `scripts.formation.states.{code}.adapter`. Raises
|
||
`ImportError` if the adapter hasn't been written yet (not every
|
||
jurisdiction has a filer).
|
||
"""
|
||
from scripts.formation.states import get_adapter
|
||
return get_adapter(self.code)
|
||
|
||
def has_adapter(self) -> bool:
|
||
"""Whether a Playwright adapter is implemented for this code."""
|
||
try:
|
||
from scripts.formation.states import get_adapter
|
||
get_adapter(self.code)
|
||
return True
|
||
except Exception:
|
||
return False
|
||
|
||
|
||
# ────────────────────────────────────────────────────────────────────── #
|
||
# Internal helpers
|
||
# ────────────────────────────────────────────────────────────────────── #
|
||
|
||
_FOREIGN_QUAL_FEE_COL = {
|
||
"llc": "foreign_llc_fee",
|
||
"pllc": "foreign_llc_fee",
|
||
"corporation": "foreign_corp_fee",
|
||
"c_corp": "foreign_corp_fee",
|
||
"s_corp": "foreign_corp_fee",
|
||
"pc": "foreign_corp_fee",
|
||
"nonprofit": "foreign_corp_fee",
|
||
}
|
||
|
||
_FORMATION_FEE_COL = {
|
||
"llc": "llc_formation_fee",
|
||
"pllc": "llc_formation_fee",
|
||
"corporation": "corp_formation_fee",
|
||
"c_corp": "corp_formation_fee",
|
||
"s_corp": "corp_formation_fee",
|
||
"pc": "corp_formation_fee",
|
||
"nonprofit": "corp_formation_fee",
|
||
}
|
||
|
||
|
||
def _normalize_entity_type(et: str) -> str:
|
||
"""Collapse variants to the canonical key used in the fee-column maps."""
|
||
return (et or "").strip().lower().replace("-", "_")
|
||
|
||
|
||
def _connect():
|
||
return psycopg2.connect(os.environ.get("DATABASE_URL", ""))
|
||
|
||
|
||
def _query_one(sql: str, params: tuple) -> Optional[dict]:
|
||
conn = _connect()
|
||
try:
|
||
with conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor) as cur:
|
||
cur.execute(sql, params)
|
||
row = cur.fetchone()
|
||
return dict(row) if row else None
|
||
finally:
|
||
conn.close()
|
||
|
||
|
||
def _row_to_config(row: dict) -> JurisdictionConfig:
|
||
entity_types_raw = row.get("entity_types_json") or []
|
||
entity_types = [
|
||
EntityTypeSpec(code=e["code"], label=e["label"])
|
||
for e in entity_types_raw
|
||
if isinstance(e, dict) and "code" in e
|
||
]
|
||
return JurisdictionConfig(
|
||
code=row["code"],
|
||
name=row["name"],
|
||
country=row["country"],
|
||
kind=row["kind"],
|
||
currency=row["currency"],
|
||
timezone=row.get("timezone"),
|
||
portal_name=row.get("portal_name"),
|
||
portal_url=row.get("portal_url"),
|
||
portal_login_required=bool(row.get("portal_login_required", False)),
|
||
entity_types=entity_types,
|
||
supports_foreign_qualification=bool(
|
||
row.get("supports_foreign_qualification", True),
|
||
),
|
||
foreign_qual_portal_url=row.get("foreign_qual_portal_url"),
|
||
foreign_qual_requires_coa=bool(row.get("foreign_qual_requires_coa", True)),
|
||
nwra_foreign_qual_wholesale_cents=row.get("nwra_foreign_qual_wholesale_cents"),
|
||
notes=row.get("notes"),
|
||
)
|
||
|
||
|
||
# ────────────────────────────────────────────────────────────────────── #
|
||
# Public API
|
||
# ────────────────────────────────────────────────────────────────────── #
|
||
|
||
|
||
@lru_cache(maxsize=128)
|
||
def get_jurisdiction(code: str) -> JurisdictionConfig:
|
||
"""Load the jurisdiction for `code` from the DB.
|
||
|
||
Cached — safe because the table is seeded once per deploy. Flush
|
||
with `get_jurisdiction.cache_clear()` if you update a row live.
|
||
"""
|
||
code_u = (code or "").strip().upper()
|
||
if not code_u:
|
||
raise ValueError("jurisdiction code required")
|
||
row = _query_one(
|
||
"""
|
||
SELECT code, name, country, kind, currency, timezone,
|
||
portal_name, portal_url, portal_login_required,
|
||
entity_types_json,
|
||
supports_foreign_qualification,
|
||
foreign_qual_portal_url, foreign_qual_requires_coa,
|
||
nwra_foreign_qual_wholesale_cents,
|
||
notes
|
||
FROM jurisdictions
|
||
WHERE code = %s
|
||
""",
|
||
(code_u,),
|
||
)
|
||
if not row:
|
||
raise ValueError(f"Unknown jurisdiction code: {code_u}")
|
||
return _row_to_config(row)
|
||
|
||
|
||
def list_jurisdictions(
|
||
country: Optional[str] = None,
|
||
kind: Optional[str] = None,
|
||
supports_foreign_qualification: Optional[bool] = None,
|
||
) -> list[JurisdictionConfig]:
|
||
"""Return every jurisdiction, optionally filtered."""
|
||
sql = """
|
||
SELECT code, name, country, kind, currency, timezone,
|
||
portal_name, portal_url, portal_login_required,
|
||
entity_types_json,
|
||
supports_foreign_qualification,
|
||
foreign_qual_portal_url, foreign_qual_requires_coa,
|
||
nwra_foreign_qual_wholesale_cents,
|
||
notes
|
||
FROM jurisdictions
|
||
"""
|
||
conditions: list[str] = []
|
||
params: list = []
|
||
if country:
|
||
conditions.append("country = %s")
|
||
params.append(country.upper())
|
||
if kind:
|
||
conditions.append("kind = %s")
|
||
params.append(kind.lower())
|
||
if supports_foreign_qualification is not None:
|
||
conditions.append("supports_foreign_qualification = %s")
|
||
params.append(supports_foreign_qualification)
|
||
if conditions:
|
||
sql += "\n WHERE " + " AND ".join(conditions)
|
||
sql += "\n ORDER BY country, code"
|
||
|
||
conn = _connect()
|
||
try:
|
||
with conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor) as cur:
|
||
cur.execute(sql, tuple(params))
|
||
return [_row_to_config(dict(r)) for r in cur.fetchall()]
|
||
finally:
|
||
conn.close()
|
||
|
||
|
||
__all__ = [
|
||
"EntityTypeSpec",
|
||
"JurisdictionConfig",
|
||
"get_jurisdiction",
|
||
"list_jurisdictions",
|
||
]
|