Writing Production-Grade Stored Procedures with Snowpark Python

Why Snowpark Stored Procedures?

Traditional Snowflake stored procedures use JavaScript — functional but limited. Snowpark Python stored procedures give you the full Python ecosystem: pandas, scikit-learn, requests, and any package on the Snowflake Anaconda channel.

Project Structure

my_sp_project/
├── procedures/
│   ├── __init__.py
│   └── my_procedure.py     # The handler function
├── deploy.py               # Deployment script
├── requirements.txt
└── .env

Writing the Handler

from snowflake.snowpark import Session
from snowflake.snowpark.functions import col
import pandas as pd

def my_procedure(session: Session, param: str) -> str:
    """
    Handler function — must accept Session as first argument.
    All other arguments map to SQL procedure parameters.
    """
    df = session.table("MY_TABLE").filter(col("STATUS") == param)
    pandas_df: pd.DataFrame = df.to_pandas()

    # Your business logic here
    result = pandas_df.groupby("CATEGORY")["AMOUNT"].sum().to_dict()

    return str(result)

Deploying via Python

from snowflake.snowpark import Session

def deploy(session: Session):
    session.sproc.register(
        func=my_procedure,
        name="MY_PROCEDURE",
        packages=["snowflake-snowpark-python", "pandas"],
        is_permanent=True,
        stage_location="@MY_STAGE",
        replace=True,
        execute_as="caller",   # or "owner"
    )

if __name__ == "__main__":
    session = Session.builder.configs({...}).create()
    deploy(session)
    print("Deployed successfully.")

Testing Locally

Test the handler without deploying — just pass a mock Session:

# test_my_procedure.py
from snowflake.snowpark import Session
from procedures.my_procedure import my_procedure

def test_locally():
    session = Session.builder.configs({
        "account": "...",
        "user": "...",
        "password": "...",
        "warehouse": "COMPUTE_WH",
        "database": "DEV_DB",
        "schema": "PUBLIC",
    }).create()

    result = my_procedure(session, "ACTIVE")
    print(result)
    session.close()

CI/CD with GitHub Actions

name: Deploy Stored Procedures
on:
  push:
    branches: [main]
    paths: ["procedures/**"]

jobs:
  deploy:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with: { python-version: "3.11" }
      - run: pip install -r requirements.txt
      - run: python deploy.py
        env:
          SNOWFLAKE_ACCOUNT: ${{ secrets.SNOWFLAKE_ACCOUNT }}
          SNOWFLAKE_USER: ${{ secrets.SNOWFLAKE_USER }}
          SNOWFLAKE_PASSWORD: ${{ secrets.SNOWFLAKE_PASSWORD }}

Conclusion

Snowpark Python stored procedures are a first-class way to bring Python logic into Snowflake. With proper project structure, local testing, and CI/CD automation, you can ship production-quality procedures with the same rigour as any Python service.