Finally fixed greenlet_spawn sqllchemy error. The horror

alembic/versions/2025_12_03_1142-3147e421bc47_removed_hashed_customer_completly.py

@@ -0,0 +1,63 @@
+"""removed hashed_customer completly
+
+Revision ID: 3147e421bc47
+Revises: 0fbeb40dbb2c
+Create Date: 2025-12-03 11:42:05.722690
+
+"""
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+from sqlalchemy.dialects import postgresql
+
+# revision identifiers, used by Alembic.
+revision: str = '3147e421bc47'
+down_revision: Union[str, Sequence[str], None] = '0fbeb40dbb2c'
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    """Upgrade schema."""
+    # ### commands auto generated by Alembic - please adjust! ###
+    
+    op.drop_index(op.f('ix_conversion_guests_hashed_customer_id'), table_name='conversion_guests')
+    op.drop_constraint(op.f('fk_conversion_guests_hashed_customer_id_hashed_customers'), 'conversion_guests', type_='foreignkey')
+    op.drop_column('conversion_guests', 'hashed_customer_id')
+    op.drop_index(op.f('ix_conversions_hashed_customer_id'), table_name='conversions')
+    op.drop_constraint(op.f('conversions_hashed_customer_id_fkey'), 'conversions', type_='foreignkey')
+    op.drop_column('conversions', 'hashed_customer_id')
+    op.drop_table('hashed_customers')
+    # ### end Alembic commands ###
+
+
+def downgrade() -> None:
+    """Downgrade schema."""
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.add_column('conversions', sa.Column('hashed_customer_id', sa.INTEGER(), autoincrement=False, nullable=True))
+    op.create_foreign_key(op.f('conversions_hashed_customer_id_fkey'), 'conversions', 'hashed_customers', ['hashed_customer_id'], ['id'])
+    op.create_index(op.f('ix_conversions_hashed_customer_id'), 'conversions', ['hashed_customer_id'], unique=False)
+    op.add_column('conversion_guests', sa.Column('hashed_customer_id', sa.INTEGER(), autoincrement=False, nullable=True))
+    op.create_foreign_key(op.f('fk_conversion_guests_hashed_customer_id_hashed_customers'), 'conversion_guests', 'hashed_customers', ['hashed_customer_id'], ['id'])
+    op.create_index(op.f('ix_conversion_guests_hashed_customer_id'), 'conversion_guests', ['hashed_customer_id'], unique=False)
+    op.create_table('hashed_customers',
+    sa.Column('id', sa.INTEGER(), autoincrement=True, nullable=False),
+    sa.Column('customer_id', sa.INTEGER(), autoincrement=False, nullable=True),
+    sa.Column('contact_id', sa.VARCHAR(), autoincrement=False, nullable=True),
+    sa.Column('hashed_email', sa.VARCHAR(length=64), autoincrement=False, nullable=True),
+    sa.Column('hashed_phone', sa.VARCHAR(length=64), autoincrement=False, nullable=True),
+    sa.Column('hashed_given_name', sa.VARCHAR(length=64), autoincrement=False, nullable=True),
+    sa.Column('hashed_surname', sa.VARCHAR(length=64), autoincrement=False, nullable=True),
+    sa.Column('hashed_city', sa.VARCHAR(length=64), autoincrement=False, nullable=True),
+    sa.Column('hashed_postal_code', sa.VARCHAR(length=64), autoincrement=False, nullable=True),
+    sa.Column('hashed_country_code', sa.VARCHAR(length=64), autoincrement=False, nullable=True),
+    sa.Column('hashed_gender', sa.VARCHAR(length=64), autoincrement=False, nullable=True),
+    sa.Column('hashed_birth_date', sa.VARCHAR(length=64), autoincrement=False, nullable=True),
+    sa.Column('created_at', postgresql.TIMESTAMP(timezone=True), autoincrement=False, nullable=True),
+    sa.ForeignKeyConstraint(['customer_id'], ['customers.id'], name=op.f('hashed_customers_customer_id_fkey'), ondelete='SET NULL'),
+    sa.PrimaryKeyConstraint('id', name=op.f('hashed_customers_pkey')),
+    sa.UniqueConstraint('contact_id', name=op.f('uq_hashed_customers_contact_id'), postgresql_include=[], postgresql_nulls_not_distinct=False),
+    sa.UniqueConstraint('customer_id', name=op.f('uq_hashed_customers_customer_id'), postgresql_include=[], postgresql_nulls_not_distinct=False)
+    )
+    # ### end Alembic commands ###

alembic/versions/2025_12_03_1225-263bed87114f_add_conversions_conversion_guests_fk.py

@@ -0,0 +1,32 @@
+"""add conversions→conversion_guests fk
+
+Revision ID: 263bed87114f
+Revises: 3147e421bc47
+Create Date: 2025-12-03 12:25:12.820232
+
+"""
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision: str = '263bed87114f'
+down_revision: Union[str, Sequence[str], None] = '3147e421bc47'
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    """Upgrade schema."""
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.create_foreign_key('fk_conversions_guest', 'conversions', 'conversion_guests', ['hotel_id', 'guest_id'], ['hotel_id', 'guest_id'])
+    # ### end Alembic commands ###
+
+
+def downgrade() -> None:
+    """Downgrade schema."""
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.drop_constraint('fk_conversions_guest', 'conversions', type_='foreignkey')
+    # ### end Alembic commands ###

src/alpine_bits_python/api.py

@@ -28,7 +28,7 @@ from fastapi.security import (
 from pydantic import BaseModel
 from slowapi.errors import RateLimitExceeded
 from sqlalchemy import and_, select, update
-from sqlalchemy.ext.asyncio import async_sessionmaker
+from sqlalchemy.ext.asyncio import async_sessionmaker, AsyncSession
 from sqlalchemy.orm import selectinload
 
 from alpine_bits_python.hotel_service import HotelService
@@ -704,7 +704,7 @@ async def validate_basic_auth(
 async def handle_webhook_unified(
     request: Request,
     webhook_secret: str,
-    db_session=Depends(get_async_session),
+    db_session: AsyncSession = Depends(get_async_session),
 ):
     """Unified webhook handler with deduplication and routing.
 
@@ -831,6 +831,9 @@ async def handle_webhook_unified(
         if not webhook_endpoint:
             raise HTTPException(status_code=404, detail="Webhook not found")
 
+    webhook_endpoint_id = webhook_endpoint.id
+    webhook_hotel_id = webhook_endpoint.hotel_id
+
     # Verify hotel is active
     if not webhook_endpoint.hotel.is_active:
         raise HTTPException(status_code=404, detail="Hotel is not active")
@@ -845,8 +848,8 @@ async def handle_webhook_unified(
 
     webhook_request_data = WebhookRequestData(
         payload_json=payload,
-        webhook_endpoint_id=webhook_endpoint.id,
-        hotel_id=webhook_endpoint.hotel_id,
+        webhook_endpoint_id=webhook_endpoint_id,
+        hotel_id=webhook_hotel_id,
         status=WebhookStatus.PROCESSING,
         processing_started_at=timestamp,
         created_at=timestamp,
@@ -908,12 +911,17 @@ async def handle_webhook_unified(
         db_session.add(webhook_request)
         await db_session.flush()
 
+    webhook_request_id = webhook_request.id
+
     try:
         # 6. Get processor for webhook_type
         processor = webhook_registry.get_processor(webhook_endpoint.webhook_type)
         if not processor:
             raise ValueError(f"No processor for type: {webhook_endpoint.webhook_type}")
 
+        # Persist the webhook row before handing off to processors
+        await db_session.commit()
+
         # 7. Process webhook with simplified interface
         result = await processor.process(
             webhook_request=webhook_request,
@@ -922,34 +930,50 @@ async def handle_webhook_unified(
             event_dispatcher=request.app.state.event_dispatcher,
         )
 
-        # 8. Update status and link created entities when available
-        webhook_request.status = WebhookStatus.COMPLETED
-        webhook_request.processing_completed_at = datetime.now(UTC)
+        if not db_session.in_transaction():
+            await db_session.begin()
 
-        created_customer_id = result.get("customer_id") if isinstance(result, dict) else None
-        created_reservation_id = (
-            result.get("reservation_id") if isinstance(result, dict) else None
+        completion_values = {
+            "status": WebhookStatus.COMPLETED,
+            "processing_completed_at": datetime.now(UTC),
+        }
+
+        if isinstance(result, dict):
+            created_customer_id = result.get("customer_id")
+            created_reservation_id = result.get("reservation_id")
+            if created_customer_id:
+                completion_values["created_customer_id"] = created_customer_id
+            if created_reservation_id:
+                completion_values["created_reservation_id"] = created_reservation_id
+
+        await db_session.execute(
+            update(WebhookRequest)
+            .where(WebhookRequest.id == webhook_request_id)
+            .values(**completion_values)
         )
-
-        if created_customer_id:
-            webhook_request.created_customer_id = created_customer_id
-        if created_reservation_id:
-            webhook_request.created_reservation_id = created_reservation_id
-
         await db_session.commit()
 
         return {
             **result,
-            "webhook_id": webhook_request.id,
-            "hotel_id": webhook_endpoint.hotel_id,
+            "webhook_id": webhook_request_id,
+            "hotel_id": webhook_hotel_id,
         }
 
     except Exception as e:
         _LOGGER.exception("Error processing webhook: %s", e)
 
-        webhook_request.status = WebhookStatus.FAILED
-        webhook_request.last_error = str(e)[:2000]
-        webhook_request.processing_completed_at = datetime.now(UTC)
+        await db_session.rollback()
+        if not db_session.in_transaction():
+            await db_session.begin()
+        await db_session.execute(
+            update(WebhookRequest)
+            .where(WebhookRequest.id == webhook_request_id)
+            .values(
+                status=WebhookStatus.FAILED,
+                last_error=str(e)[:2000],
+                processing_completed_at=datetime.now(UTC),
+            )
+        )
         await db_session.commit()
 
         raise HTTPException(status_code=500, detail="Error processing webhook")

src/alpine_bits_python/conversion_service.py

@@ -471,7 +471,6 @@ class ConversionService:
             "total_daily_sales": 0,
             "matched_to_reservation": 0,
             "matched_to_customer": 0,
-            "matched_to_hashed_customer": 0,
             "unmatched": 0,
             "errors": 0,
         }
@@ -629,10 +628,8 @@ class ConversionService:
             from sqlalchemy.orm import selectinload
 
             query = select(Reservation).options(
-                selectinload(Reservation.customer).selectinload(
-                    Customer.hashed_version
-                ),
-                selectinload(Reservation.hashed_customer),
+                selectinload(Reservation.customer),
+                
             )
             result = await session.execute(query)
             reservations = result.scalars().all()
@@ -645,13 +642,11 @@ class ConversionService:
                 if hotel_code not in self._reservation_cache:
                     self._reservation_cache[hotel_code] = []
                 # Cache the hashed customer - prefer direct relationship, fall back to customer relationship
-                hashed_customer = None
-                if reservation.hashed_customer:
-                    hashed_customer = reservation.hashed_customer
-                elif reservation.customer and reservation.customer.hashed_version:
-                    hashed_customer = reservation.customer.hashed_version
+                customer = None
+                if reservation.customer:
+                    customer = reservation.customer
                 self._reservation_cache[hotel_code].append(
-                    (reservation, hashed_customer)
+                    (reservation, customer)
                 )
 
             self._cache_initialized = True
@@ -1431,7 +1426,6 @@ class ConversionService:
 
                     conversion.reservation_id = matched_reservation.id
                     conversion.customer_id = matched_hashed_customer.id
-                    conversion.hashed_customer_id = matched_hashed_customer.id
                     conversion.directly_attributable = True
                     conversion.guest_matched = True
                     conversion.updated_at = datetime.now()
@@ -1447,7 +1441,6 @@ class ConversionService:
                 elif matched_hashed_customer and conversion.customer_id is None:
                     # Only count new customer matches (conversions that didn't have a customer before)
                     conversion.customer_id = matched_hashed_customer.id
-                    conversion.hashed_customer_id = matched_hashed_customer.id
                     conversion.directly_attributable = False
                     conversion.guest_matched = True
                     conversion.updated_at = datetime.now()
@@ -1477,41 +1470,40 @@ class ConversionService:
             session: AsyncSession for database queries
 
         """
-        # Get all ConversionGuests that have ANY customer link
-        # This includes:
-        # 1. Guests matched via guest-details (hashed_customer_id is not null)
-        # 2. Guests matched via ID-based matching (customer_id is not null via conversion)
+        # Collect every guest/customer pair derived from conversions.
         result = await session.execute(
-            select(ConversionGuest).where(
-                ConversionGuest.hashed_customer_id.isnot(None)
+            select(Conversion.guest_id, Conversion.customer_id).where(
+                Conversion.guest_id.isnot(None), Conversion.customer_id.isnot(None)
             )
         )
-        matched_guests = result.scalars().all()
+        guest_customer_rows = result.all()
 
-        if not matched_guests:
+        if not guest_customer_rows:
             _LOGGER.debug("Phase 3d: No matched guests to check for regularity")
             return
 
+        # Deduplicate by guest_id to avoid recalculating when multiple conversions share the same guest.
+        guest_to_customer: dict[int, int] = {}
+        for guest_id, customer_id in guest_customer_rows:
+            if guest_id is None or customer_id is None:
+                continue
+            if guest_id not in guest_to_customer:
+                guest_to_customer[guest_id] = customer_id
+            elif guest_to_customer[guest_id] != customer_id:
+                _LOGGER.warning(
+                    "Guest %s linked to multiple customers (%s, %s); keeping first match",
+                    guest_id,
+                    guest_to_customer[guest_id],
+                    customer_id,
+                )
+
         _LOGGER.debug(
-            "Phase 3d: Checking regularity for %d matched guests", len(matched_guests)
+            "Phase 3d: Checking regularity for %d matched guests",
+            len(guest_to_customer),
         )
 
-        for conversion_guest in matched_guests:
-            if not conversion_guest.hashed_customer_id:
-                continue
-
-            # Get the customer ID from the hashed_customer
-            hashed_customer_result = await session.execute(
-                select(Customer).where(
-                    Customer.id == conversion_guest.hashed_customer_id
-                )
-            )
-            hashed_customer = hashed_customer_result.scalar_one_or_none()
-
-            if hashed_customer and hashed_customer.id:
-                await self._check_if_guest_is_regular(
-                    conversion_guest.guest_id, hashed_customer.id, session
-                )
+        for guest_id, customer_id in guest_to_customer.items():
+            await self._check_if_guest_is_regular(guest_id, customer_id, session)
 
     async def _match_conversions_from_db_sequential(
         self, pms_reservation_ids: list[str], stats: dict[str, int]
@@ -1727,7 +1719,6 @@ class ConversionService:
         # Guest detail matching is deferred to Phase 3b/3c
         matched_reservation = None
         matched_customer = None
-        matched_hashed_customer = None
 
         if conversion.advertising_campagne:
             matched_reservation = await self._match_by_advertising(
@@ -1742,8 +1733,6 @@ class ConversionService:
 
             if matched_reservation:
                 matched_customer = matched_reservation.customer
-                if matched_customer and matched_customer.hashed_version:
-                    matched_hashed_customer = matched_customer.hashed_version
 
                 _LOGGER.info(
                     "Phase 3a: Matched conversion by advertising ID (pms_id=%s, reservation_id=%d)",
@@ -1752,23 +1741,17 @@ class ConversionService:
                 )
 
         # Update the conversion with matched entities if found
-        if matched_reservation or matched_customer or matched_hashed_customer:
+        if matched_reservation or matched_customer:
             conversion.reservation_id = (
                 matched_reservation.id if matched_reservation else None
             )
             conversion.customer_id = matched_customer.id if matched_customer else None
-            conversion.hashed_customer_id = (
-                matched_hashed_customer.id if matched_hashed_customer else None
-            )
+
 
             # ID-based matches are always directly attributable
             conversion.directly_attributable = True
             conversion.guest_matched = False
 
-            # Update conversion_guest with hashed_customer reference if matched
-            if conversion_guest and matched_hashed_customer:
-                conversion_guest.hashed_customer_id = matched_hashed_customer.id
-
             conversion.updated_at = datetime.now()
 
         # Update stats if provided
@@ -1777,8 +1760,6 @@ class ConversionService:
                 stats["matched_to_reservation"] += 1
             elif matched_customer:
                 stats["matched_to_customer"] += 1
-            elif matched_hashed_customer:
-                stats["matched_to_hashed_customer"] += 1
             else:
                 stats["unmatched"] += 1
 

src/alpine_bits_python/customer_service.py

@@ -6,7 +6,7 @@ from pydantic import ValidationError
 from sqlalchemy import select
 from sqlalchemy.ext.asyncio import AsyncSession
 
-from .db import Customer, HashedCustomer
+from .db import Customer
 from .logging_config import get_logger
 from .schemas import CustomerData
 

src/alpine_bits_python/db.py

@@ -362,36 +362,6 @@ class Customer(Base):
         self.hashed_birth_date = self._normalize_and_hash(self.birth_date)
 
 
-class HashedCustomer(Base):
-    """Hashed customer data for Meta Conversion API.
-
-    Stores SHA256 hashed versions of customer PII according to Meta's requirements.
-    This allows sending conversion events without exposing raw customer data.
-    """
-
-    __tablename__ = "hashed_customers"
-    id = Column(Integer, primary_key=True)
-    customer_id = Column(
-        Integer,
-        ForeignKey("customers.id", ondelete="SET NULL"),
-        unique=True,
-        nullable=True,
-    )
-    contact_id = Column(String, unique=True)  # Keep unhashed for reference
-    hashed_email = Column(String(64))  # SHA256 produces 64 hex chars
-    hashed_phone = Column(String(64))
-    hashed_given_name = Column(String(64))
-    hashed_surname = Column(String(64))
-    hashed_city = Column(String(64))
-    hashed_postal_code = Column(String(64))
-    hashed_country_code = Column(String(64))
-    hashed_gender = Column(String(64))
-    hashed_birth_date = Column(String(64))
-    created_at = Column(DateTime(timezone=True))
-
-    customer = relationship(
-        "Customer", backref=backref("hashed_version", uselist=False, lazy="joined")
-    )
 
 
 class ConversionGuest(Base):
@@ -430,10 +400,7 @@ class ConversionGuest(Base):
     hashed_country_code = Column(String(64))
     hashed_birth_date = Column(String(64))
 
-    # Matched customer reference (nullable, filled after matching)
-    hashed_customer_id = Column(
-        Integer, ForeignKey("hashed_customers.id"), nullable=True, index=True
-    )
+
 
     # Guest classification
     is_regular = Column(
@@ -452,7 +419,6 @@ class ConversionGuest(Base):
         primaryjoin="and_(ConversionGuest.hotel_id == foreign(Conversion.hotel_id), "
                     "ConversionGuest.guest_id == foreign(Conversion.guest_id))",
     )
-    hashed_customer = relationship("HashedCustomer", backref="conversion_guests")
 
     @staticmethod
     def _normalize_and_hash(value):
@@ -613,9 +579,6 @@ class Conversion(Base):
         Integer, ForeignKey("reservations.id"), nullable=True, index=True
     )
     customer_id = Column(Integer, ForeignKey("customers.id"), nullable=True, index=True)
-    hashed_customer_id = Column(
-        Integer, ForeignKey("hashed_customers.id"), nullable=True, index=True
-    )
 
     # Reservation metadata from XML
     hotel_id = Column(
@@ -659,18 +622,20 @@ class Conversion(Base):
     updated_at = Column(DateTime(timezone=True))  # When this record was last updated
 
     # Table constraints
-    # Note: The relationship to ConversionGuest is handled via SQLAlchemy ORM
-    # by matching (hotel_id, guest_id) pairs, no DB-level FK constraint needed
     __table_args__ = (
         UniqueConstraint(
             "hotel_id", "pms_reservation_id", name="uq_conversion_hotel_reservation"
         ),
+        ForeignKeyConstraint(
+            ["hotel_id", "guest_id"],
+            ["conversion_guests.hotel_id", "conversion_guests.guest_id"],
+            name="fk_conversions_guest",
+        ),
     )
 
     # Relationships
     reservation = relationship("Reservation", backref="conversions")
     customer = relationship("Customer", backref="conversions")
-    hashed_customer = relationship("HashedCustomer", backref="conversions")
     guest = relationship(
         "ConversionGuest",
         back_populates="conversions",

src/alpine_bits_python/reservation_service.py

@@ -7,7 +7,7 @@ from typing import Optional
 from sqlalchemy import and_, select
 from sqlalchemy.ext.asyncio import AsyncSession
 
-from .db import AckedRequest, Customer, HashedCustomer, Reservation
+from .db import AckedRequest, Customer, Reservation
 from .schemas import ReservationData
 
 
@@ -64,17 +64,6 @@ class ReservationService:
             reservation_data, customer_id
         )
 
-        # Automatically populate hashed_customer_id from the customer
-        # Since hashed_customer is always created when a customer is created,
-        # we can get it by querying for the hashed_customer with matching customer_id
-        hashed_customer_result = await self.session.execute(
-            select(HashedCustomer).where(
-                HashedCustomer.customer_id == customer_id
-            )
-        )
-        hashed_customer = hashed_customer_result.scalar_one_or_none()
-        if hashed_customer:
-            reservation.hashed_customer_id = hashed_customer.id
 
         self.session.add(reservation)
 

src/alpine_bits_python/schemas.py

@@ -562,7 +562,6 @@ class ConversionData(BaseModel):
     # Foreign key references (nullable - matched after creation)
     reservation_id: int | None = Field(None, gt=0)
     customer_id: int | None = Field(None, gt=0)
-    hashed_customer_id: int | None = Field(None, gt=0)
 
     # Required reservation metadata from PMS
     hotel_id: str = Field(..., min_length=1, max_length=50)
@@ -591,7 +590,7 @@ class ConversionData(BaseModel):
 
     @field_validator(
         "pms_reservation_id", "guest_id", "reservation_id", "customer_id",
-        "hashed_customer_id", mode="before"
+        mode="before"
     )
     @classmethod
     def convert_int_fields(cls, v: Any) -> int | None:

src/alpine_bits_python/util/migrate_sqlite_to_postgres.py

@@ -51,7 +51,6 @@ from alpine_bits_python.db import (
     AckedRequest,
     Base,
     Customer,
-    HashedCustomer,
     Reservation,
     get_database_url,
 )

src/alpine_bits_python/webhook_processor.py

@@ -203,7 +203,7 @@ async def process_wix_form_submission(
         "name_title": None,
     }
 
-    # This automatically creates/updates both Customer and HashedCustomer
+    # This automatically creates/updates Customer
     db_customer = await customer_service.get_or_create_customer(customer_data)
 
     # Determine hotel_code and hotel_name

tests/test_customer_service.py

@@ -6,7 +6,7 @@ from sqlalchemy import select
 from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
 
 from alpine_bits_python.customer_service import CustomerService
-from alpine_bits_python.db import Base, Customer, HashedCustomer
+from alpine_bits_python.db import Base, Customer
 
 
 @pytest_asyncio.fixture