megaproxy/MyMidas
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions
MyMidas/backend/alembic/versions/0006_add_tax_tables.py
megaproxy afb5e99bb2 Add recurring transaction detection, subscriptions page, and UK tax reporting 
- Recurring service: auto-detects direct debits/subscriptions from CSV imports
  using frequency analysis; manual toggle in transaction detail drawer
- Subscriptions page (/subscriptions): groups recurring payments with monthly
  cost equivalents, next-payment badges, and re-scan trigger
- UK Tax page (/tax): payslips/P60 entry, income tax + NI + CGT + dividend tax
  calculations, configurable rate tables per tax year (pre-seeded 2024/25 and
  2025/26), editable in-app so Budget changes need no rebuild
- Migration 0006: tax_rate_configs, tax_profiles, payslips, manual_cgt_disposals
  with RLS; seeds 2025/2026 rate configs for existing users
- Chart tooltip fix: all Recharts tooltips now use TOOLTIP_STYLE constant so
  they render correctly across all dark/light themes

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-04-23 21:40:02 +00:00

186 lines
7.7 KiB
Python
Raw Blame History

"""add tax tables
Revision ID: 0006
Revises: 0005
Create Date: 2026-04-23
"""
import uuid
from alembic import op
import sqlalchemy as sa
from sqlalchemy.dialects import postgresql
revision = "0006"
down_revision = "0005"
branch_labels = None
depends_on = None
# ---------------------------------------------------------------------------
# Seed data for 2025 and 2026
# ---------------------------------------------------------------------------
_INCOME_TAX_BANDS_2025 = {
"bands": [
{"from": 0, "to": 12570, "rate": 0.00},
{"from": 12570, "to": 50270, "rate": 0.20},
{"from": 50270, "to": 125140, "rate": 0.40},
{"from": 125140, "to": None, "rate": 0.45},
]
}
_NI_BANDS_2025 = {
"bands": [
{"from": 0, "to": 12570, "rate": 0.00},
{"from": 12570, "to": 50270, "rate": 0.08},
{"from": 50270, "to": None, "rate": 0.02},
]
}
_CGT_2025 = {"exempt": 3000, "basic_rate": 0.18, "higher_rate": 0.24}
_DIVIDEND_2025 = {
"allowance": 500,
"basic_rate": 0.0875,
"higher_rate": 0.3375,
"additional_rate": 0.3935,
}
# 2026 thresholds remain frozen; rates unchanged from 2025
_SEED = {
2025: {
"income_tax": _INCOME_TAX_BANDS_2025,
"ni": _NI_BANDS_2025,
"cgt": _CGT_2025,
"dividend": _DIVIDEND_2025,
},
2026: {
"income_tax": _INCOME_TAX_BANDS_2025,
"ni": _NI_BANDS_2025,
"cgt": _CGT_2025,
"dividend": _DIVIDEND_2025,
},
}
def upgrade() -> None:
# ------------------------------------------------------------------
# tax_rate_configs
# ------------------------------------------------------------------
op.create_table(
"tax_rate_configs",
sa.Column("id", postgresql.UUID(as_uuid=True), primary_key=True),
sa.Column("user_id", postgresql.UUID(as_uuid=True),
sa.ForeignKey("users.id", ondelete="CASCADE"), nullable=False),
sa.Column("tax_year", sa.Integer, nullable=False),
sa.Column("rate_type", sa.String(30), nullable=False),
sa.Column("config", postgresql.JSONB, nullable=False),
sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False),
)
op.create_unique_constraint(
"uq_tax_rate_configs_user_year_type",
"tax_rate_configs",
["user_id", "tax_year", "rate_type"],
)
op.create_index("ix_tax_rate_configs_user_id", "tax_rate_configs", ["user_id"])
# ------------------------------------------------------------------
# tax_profiles
# ------------------------------------------------------------------
op.create_table(
"tax_profiles",
sa.Column("id", postgresql.UUID(as_uuid=True), primary_key=True),
sa.Column("user_id", postgresql.UUID(as_uuid=True),
sa.ForeignKey("users.id", ondelete="CASCADE"), nullable=False),
sa.Column("tax_year", sa.Integer, nullable=False),
sa.Column("employer_name_enc", sa.LargeBinary, nullable=True),
sa.Column("tax_code", sa.String(20), nullable=False, server_default="1257L"),
sa.Column("is_cumulative", sa.Boolean, nullable=False, server_default="true"),
sa.Column("created_at", sa.DateTime(timezone=True), nullable=False),
sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False),
)
op.create_unique_constraint(
"uq_tax_profiles_user_year",
"tax_profiles",
["user_id", "tax_year"],
)
op.create_index("ix_tax_profiles_user_id", "tax_profiles", ["user_id"])
# ------------------------------------------------------------------
# payslips
# ------------------------------------------------------------------
op.create_table(
"payslips",
sa.Column("id", postgresql.UUID(as_uuid=True), primary_key=True),
sa.Column("user_id", postgresql.UUID(as_uuid=True),
sa.ForeignKey("users.id", ondelete="CASCADE"), nullable=False),
sa.Column("tax_profile_id", postgresql.UUID(as_uuid=True),
sa.ForeignKey("tax_profiles.id", ondelete="CASCADE"), nullable=False),
sa.Column("period_month", sa.SmallInteger, nullable=True),
sa.Column("period_year", sa.SmallInteger, nullable=False),
sa.Column("gross_pay", sa.Numeric(14, 2), nullable=False),
sa.Column("income_tax_withheld", sa.Numeric(14, 2), nullable=False),
sa.Column("ni_withheld", sa.Numeric(14, 2), nullable=False),
sa.Column("net_pay", sa.Numeric(14, 2), nullable=False),
sa.Column("is_p60", sa.Boolean, nullable=False, server_default="false"),
sa.Column("notes_enc", sa.LargeBinary, nullable=True),
sa.Column("created_at", sa.DateTime(timezone=True), nullable=False),
)
op.create_index("ix_payslips_user_id", "payslips", ["user_id"])
op.create_index("ix_payslips_tax_profile_id", "payslips", ["tax_profile_id"])
# ------------------------------------------------------------------
# manual_cgt_disposals
# ------------------------------------------------------------------
op.create_table(
"manual_cgt_disposals",
sa.Column("id", postgresql.UUID(as_uuid=True), primary_key=True),
sa.Column("user_id", postgresql.UUID(as_uuid=True),
sa.ForeignKey("users.id", ondelete="CASCADE"), nullable=False),
sa.Column("tax_year", sa.Integer, nullable=False),
sa.Column("disposal_date", sa.Date, nullable=False),
sa.Column("asset_description_enc", sa.LargeBinary, nullable=False),
sa.Column("proceeds", sa.Numeric(14, 2), nullable=False),
sa.Column("cost_basis", sa.Numeric(14, 2), nullable=False),
sa.Column("notes_enc", sa.LargeBinary, nullable=True),
sa.Column("created_at", sa.DateTime(timezone=True), nullable=False),
)
op.create_index("ix_manual_cgt_disposals_user_id", "manual_cgt_disposals", ["user_id"])
# ------------------------------------------------------------------
# RLS
# ------------------------------------------------------------------
for table in ["tax_rate_configs", "tax_profiles", "payslips", "manual_cgt_disposals"]:
op.execute(f"ALTER TABLE {table} ENABLE ROW LEVEL SECURITY")
op.execute(f"""
CREATE POLICY {table}_user_isolation ON {table}
USING (user_id = current_app_user_id())
""")
# ------------------------------------------------------------------
# Seed 2025 + 2026 rate configs for all existing users
# ------------------------------------------------------------------
import json
from datetime import datetime, timezone
now = datetime.now(timezone.utc).isoformat()
for tax_year, rate_types in _SEED.items():
for rate_type, config in rate_types.items():
op.execute(sa.text("""
INSERT INTO tax_rate_configs (id, user_id, tax_year, rate_type, config, updated_at)
SELECT gen_random_uuid(), id, :tax_year, :rate_type, CAST(:config AS jsonb), CAST(:updated_at AS timestamptz)
FROM users
WHERE deleted_at IS NULL
ON CONFLICT (user_id, tax_year, rate_type) DO NOTHING
""").bindparams(
tax_year=tax_year,
rate_type=rate_type,
config=json.dumps(config),
updated_at=now,
))
def downgrade() -> None:
for table in ["tax_rate_configs", "tax_profiles", "payslips", "manual_cgt_disposals"]:
op.execute(f"DROP POLICY IF EXISTS {table}_user_isolation ON {table}")
op.execute(f"ALTER TABLE {table} DISABLE ROW LEVEL SECURITY")
op.drop_table("manual_cgt_disposals")
op.drop_table("payslips")
op.drop_table("tax_profiles")
op.drop_table("tax_rate_configs")
Reference in a new issue View git blame Copy permalink