SQLAlchemy ORM Expert
Comprehensive SQLAlchemy skill for customer support tech enablement, covering ORM patterns, session management, query optimization, async operations, and PostgreSQL integration
What this skill does
# SQLAlchemy ORM Expert Skill
## Overview
This skill provides comprehensive guidance for using SQLAlchemy 2.0+ in customer support systems, focusing on ORM patterns, session management, query optimization, async operations with FastAPI, and PostgreSQL integration. It covers everything from basic model definitions to advanced patterns for high-performance support applications.
## Core Competencies
### 1. Customer Support Data Models
When building customer support systems, you need robust data models that represent tickets, users, comments, attachments, and their relationships. SQLAlchemy's declarative mapping with type hints provides a clean, modern approach.
**Base Model Setup:**
```python
from datetime import datetime
from typing import Optional, List
from sqlalchemy import String, Integer, DateTime, Text, ForeignKey, Enum, Boolean
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship
from sqlalchemy.sql import func
import enum
class Base(DeclarativeBase):
"""Base class for all ORM models"""
pass
class TicketStatus(enum.Enum):
"""Ticket status enumeration"""
OPEN = "open"
IN_PROGRESS = "in_progress"
WAITING_ON_CUSTOMER = "waiting_on_customer"
RESOLVED = "resolved"
CLOSED = "closed"
class TicketPriority(enum.Enum):
"""Ticket priority levels"""
LOW = "low"
MEDIUM = "medium"
HIGH = "high"
URGENT = "urgent"
```
**User Model:**
```python
class User(Base):
__tablename__ = "users"
id: Mapped[int] = mapped_column(primary_key=True)
email: Mapped[str] = mapped_column(String(255), unique=True, nullable=False, index=True)
full_name: Mapped[str] = mapped_column(String(255), nullable=False)
is_active: Mapped[bool] = mapped_column(Boolean, default=True, nullable=False)
is_staff: Mapped[bool] = mapped_column(Boolean, default=False, nullable=False)
password_hash: Mapped[str] = mapped_column(String(255), nullable=False)
# Timestamps
created_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True),
server_default=func.now(),
nullable=False
)
updated_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True),
server_default=func.now(),
onupdate=func.now(),
nullable=False
)
last_login: Mapped[Optional[datetime]] = mapped_column(DateTime(timezone=True))
# Relationships
tickets_created: Mapped[List["Ticket"]] = relationship(
"Ticket",
back_populates="creator",
foreign_keys="Ticket.creator_id",
cascade="all, delete-orphan"
)
tickets_assigned: Mapped[List["Ticket"]] = relationship(
"Ticket",
back_populates="assignee",
foreign_keys="Ticket.assignee_id"
)
comments: Mapped[List["Comment"]] = relationship(
"Comment",
back_populates="author",
cascade="all, delete-orphan"
)
def __repr__(self) -> str:
return f"<User(id={self.id}, email='{self.email}', name='{self.full_name}')>"
```
**Ticket Model:**
```python
class Ticket(Base):
__tablename__ = "tickets"
id: Mapped[int] = mapped_column(primary_key=True)
ticket_number: Mapped[str] = mapped_column(String(50), unique=True, nullable=False, index=True)
title: Mapped[str] = mapped_column(String(500), nullable=False)
description: Mapped[str] = mapped_column(Text, nullable=False)
# Status and priority
status: Mapped[TicketStatus] = mapped_column(
Enum(TicketStatus),
default=TicketStatus.OPEN,
nullable=False,
index=True
)
priority: Mapped[TicketPriority] = mapped_column(
Enum(TicketPriority),
default=TicketPriority.MEDIUM,
nullable=False,
index=True
)
# Foreign keys
creator_id: Mapped[int] = mapped_column(ForeignKey("users.id"), nullable=False, index=True)
assignee_id: Mapped[Optional[int]] = mapped_column(ForeignKey("users.id"), index=True)
# Soft delete
deleted_at: Mapped[Optional[datetime]] = mapped_column(DateTime(timezone=True))
# Timestamps
created_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True),
server_default=func.now(),
nullable=False,
index=True
)
updated_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True),
server_default=func.now(),
onupdate=func.now(),
nullable=False
)
resolved_at: Mapped[Optional[datetime]] = mapped_column(DateTime(timezone=True))
# Relationships
creator: Mapped["User"] = relationship(
"User",
back_populates="tickets_created",
foreign_keys=[creator_id]
)
assignee: Mapped[Optional["User"]] = relationship(
"User",
back_populates="tickets_assigned",
foreign_keys=[assignee_id]
)
comments: Mapped[List["Comment"]] = relationship(
"Comment",
back_populates="ticket",
cascade="all, delete-orphan",
order_by="Comment.created_at"
)
attachments: Mapped[List["Attachment"]] = relationship(
"Attachment",
back_populates="ticket",
cascade="all, delete-orphan"
)
tags: Mapped[List["Tag"]] = relationship(
"Tag",
secondary="ticket_tags",
back_populates="tickets"
)
def __repr__(self) -> str:
return f"<Ticket(id={self.id}, number='{self.ticket_number}', status={self.status.value})>"
```
### 2. Relationship Patterns
**One-to-Many (Comments on Tickets):**
```python
class Comment(Base):
__tablename__ = "comments"
id: Mapped[int] = mapped_column(primary_key=True)
content: Mapped[str] = mapped_column(Text, nullable=False)
is_internal: Mapped[bool] = mapped_column(Boolean, default=False, nullable=False)
# Foreign keys
ticket_id: Mapped[int] = mapped_column(ForeignKey("tickets.id"), nullable=False, index=True)
author_id: Mapped[int] = mapped_column(ForeignKey("users.id"), nullable=False, index=True)
# Timestamps
created_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True),
server_default=func.now(),
nullable=False
)
updated_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True),
server_default=func.now(),
onupdate=func.now(),
nullable=False
)
# Relationships
ticket: Mapped["Ticket"] = relationship("Ticket", back_populates="comments")
author: Mapped["User"] = relationship("User", back_populates="comments")
def __repr__(self) -> str:
return f"<Comment(id={self.id}, ticket_id={self.ticket_id}, author_id={self.author_id})>"
```
**Many-to-Many (Tags on Tickets):**
```python
from sqlalchemy import Table, Column
# Association table for many-to-many relationship
ticket_tags = Table(
"ticket_tags",
Base.metadata,
Column("ticket_id", ForeignKey("tickets.id"), primary_key=True),
Column("tag_id", ForeignKey("tags.id"), primary_key=True),
Column("created_at", DateTime(timezone=True), server_default=func.now())
)
class Tag(Base):
__tablename__ = "tags"
id: Mapped[int] = mapped_column(primary_key=True)
name: Mapped[str] = mapped_column(String(100), unique=True, nullable=False, index=True)
color: Mapped[str] = mapped_column(String(7), nullable=False) # Hex color
description: Mapped[Optional[str]] = mapped_column(String(500))
# Timestamps
created_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True),
server_default=func.now(),
nullable=False
)
# Relationships
tickets: Mapped[List["Ticket"]] = relationship(
"Ticket",
secondary=ticket_tags,
back_populates="tags"
)
def __repr__(self) -> str:
return f"<Tag(id={self.id}, name='{self.name}')>"
```
### 3. Session Management
**Synchronous Session Setup:**
```python
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, Session
from contextlib import conteRelated in Sales & CRM
process-mapper
IncludedUse when a BizOps lead, COO, or process-improvement owner needs to document an end-to-end business process (procurement, employee onboarding, incident handoff, customer-onboarding, claims adjudication) in BPMN-style notation, measure cycle times by stage, surface where work spends most of its time waiting vs. being worked, and quantify the gap between processing time and total elapsed time. Pairs Lean / Six Sigma / Theory-of-Constraints canon with deterministic stdlib-only Python tools to produce a process map, a ranked bottleneck list (with severity + root-cause hypothesis), and a cycle-time analysis (P50, P90, value-add ratio, Little's-Law throughput). Distinct from sales-pipeline, system-reliability (SLO), and strategic-OKR work — this is tactical process documentation for internal operations.
payment-integration
IncludedIntegrate payments with SePay (VietQR), Polar, Stripe, Paddle (MoR subscriptions), Creem.io (licensing). Checkout, webhooks, subscriptions, QR codes, multi-provider orders.
customer-success-manager
IncludedMonitors customer health, predicts churn risk, and identifies expansion opportunities using weighted scoring models for SaaS customer success
sales-engineer
IncludedAnalyzes RFP/RFI responses for coverage gaps, builds competitive feature comparison matrices, and plans proof-of-concept (POC) engagements for pre-sales engineering. Use when responding to RFPs, bids, or proposal requests; comparing product features against competitors; planning or scoring a customer POC or sales demo; preparing a technical proposal; or performing win/loss competitor analysis. Handles tasks described as 'RFP response', 'bid response', 'proposal response', 'competitor comparison', 'feature matrix', 'POC planning', 'sales demo prep', or 'pre-sales engineering'.
customer-success-manager
IncludedMonitors customer health, predicts churn risk, and identifies expansion opportunities using weighted scoring models for SaaS customer success
sales-engineer
IncludedAnalyzes RFP/RFI responses for coverage gaps, builds competitive feature comparison matrices, and plans proof-of-concept (POC) engagements for pre-sales engineering. Use when responding to RFPs, bids, or proposal requests; comparing product features against competitors; planning or scoring a customer POC or sales demo; preparing a technical proposal; or performing win/loss competitor analysis. Handles tasks described as 'RFP response', 'bid response', 'proposal response', 'competitor comparison', 'feature matrix', 'POC planning', 'sales demo prep', or 'pre-sales engineering'.