### Start Fakesnow Server and Connect with Snowflake Connector

Source: https://context7.com/tekumara/fakesnow/llms.txt

Demonstrates how to start a Fakesnow server within a context manager, retrieve connection arguments, and then use these arguments to establish a connection with the Snowflake connector. It includes examples of creating databases, schemas, tables, inserting data, and performing a sum aggregation.

```python
import fakesnow
import snowflake.connector

# Start server in a context manager
with fakesnow.server() as conn_kwargs:
    # conn_kwargs contains: user, password, account, host, port, protocol
    print(f"Server running on port {conn_kwargs['port']}")

    # Connect using the provided kwargs
    with snowflake.connector.connect(**conn_kwargs) as conn:
        cursor = conn.cursor()

        cursor.execute("CREATE DATABASE testdb")
        cursor.execute("USE DATABASE testdb")
        cursor.execute("CREATE SCHEMA analytics")
        cursor.execute("USE SCHEMA analytics")

        cursor.execute("CREATE TABLE sales (id INT, amount DECIMAL(10,2))")
        cursor.execute("INSERT INTO sales VALUES (1, 99.99), (2, 149.50)")

        cursor.execute("SELECT SUM(amount) FROM sales")
        print(cursor.fetchone())  # Output: (Decimal('249.49'),)
```

--------------------------------

### fakesnow.server()

Source: https://context7.com/tekumara/fakesnow/llms.txt

Starts a local mock Snowflake server instance, returning connection parameters required to connect via the standard Snowflake connector.

```APIDOC
## fakesnow.server()

### Description
Starts an in-memory Snowflake server within a context manager. It provides the necessary credentials (user, password, account, host, port, protocol) to connect using the standard snowflake-connector-python.

### Method
N/A (Python Context Manager)

### Parameters
#### Optional Parameters
- **port** (int) - Optional - The port to bind the server to.
- **session_parameters** (dict) - Optional - Configuration for database persistence (e.g., FAKESNOW_DB_PATH).

### Request Example
with fakesnow.server(port=18080) as conn_kwargs:
    # Use conn_kwargs to connect

### Response
#### Success Response
- **conn_kwargs** (dict) - Dictionary containing connection parameters: user, password, account, host, port, protocol.
```

--------------------------------

### Fakesnow CLI Usage for Running Scripts and Starting Server

Source: https://context7.com/tekumara/fakesnow/llms.txt

Provides examples of using the Fakesnow command-line interface (CLI) to run Python scripts or modules with Snowflake patching enabled. It covers options for database persistence, starting the server, and specifying host and port configurations.

```bash
# Run a script with patching
fakesnow script.py

# Run a module (e.g., pytest)
fakesnow -m pytest

# Run with database persistence
fakesnow -d ./databases script.py

# Start as HTTP server
fakesnow -s

# Start server on specific port
fakesnow -s -p 18080

# Start server on specific host and port
fakesnow -s --host 0.0.0.0 -p 18080
```

--------------------------------

### Install fakesnow

Source: https://github.com/tekumara/fakesnow/blob/main/README.md

Installation commands for fakesnow using pip, including the optional server component.

```shell
pip install fakesnow
pip install fakesnow[server]
```

--------------------------------

### Run fakesnow as a Server

Source: https://github.com/tekumara/fakesnow/blob/main/README.md

Starts a fakesnow HTTP server within a Python context manager, providing connection parameters to the client. This is ideal for testing scenarios involving subprocesses or non-Python clients.

```python
import fakesnow
import snowflake.connector

with fakesnow.server() as conn_kwargs:
    with snowflake.connector.connect(**conn_kwargs) as conn:
        print(conn.cursor().execute("SELECT 'Hello fake server!'").fetchone())
```

--------------------------------

### Mock SQLAlchemy Engine with Fakesnow (Python)

Source: https://github.com/tekumara/fakesnow/blob/main/notebooks/examples/fakesnow.ipynb

This example illustrates mocking a SQLAlchemy engine with Fakesnow for use with Snowflake. It configures the SQLAlchemy URL, creates an engine, and sets up the SQL Magic extension for Jupyter notebooks. It also shows how Fakesnow intercepts SQL execution.

```python
from snowflake.sqlalchemy import URL
from sqlalchemy import create_engine

url = URL(account="fakesnow",database="db1",schema="schema1")
engine = create_engine(url)

%reload_ext sql

%config SqlMagic.feedback = False
%config SqlMagic.displaycon = False
%config SqlMagic.autopandas = True
%sql engine
```

```sql
SELECT CURRENT_DATABASE(), CURRENT_SCHEMA();
ROLLBACK;
```

```sql
SELECT 'Hello fake world!'
```

--------------------------------

### Setup Fakesnow pytest Fixtures

Source: https://github.com/tekumara/fakesnow/blob/main/README.md

Configures pytest to use Fakesnow fixtures by adding the plugin to conftest.py. This enables automatic patching of standard Snowflake imports for testing.

```python
pytest_plugins = "fakesnow.fixtures"

@pytest.fixture(scope="session", autouse=True)
def setup(_fakesnow_session: None) -> Iterator[None]:
    yield
```

--------------------------------

### Integrate fakesnow with pytest

Source: https://context7.com/tekumara/fakesnow/llms.txt

Provides examples of using built-in pytest fixtures to automatically patch Snowflake connections during testing. Includes both session-wide patching and server-based testing.

```python
# conftest.py
from typing import Iterator
import pytest

pytest_plugins = "fakesnow.fixtures"

@pytest.fixture(scope="session", autouse=True)
def setup(_fakesnow_session: None) -> Iterator[None]:
    yield

# test_example.py
import snowflake.connector

def test_basic_query():
    conn = snowflake.connector.connect(database="testdb", schema="public")
    cursor = conn.cursor()
    cursor.execute("CREATE TABLE test_table (id INT, value VARCHAR)")
    cursor.execute("INSERT INTO test_table VALUES (1, 'test')")
    cursor.execute("SELECT * FROM test_table")
    assert cursor.fetchone() == (1, "test")

def test_with_server(fakesnow_server: dict):
    with snowflake.connector.connect(**fakesnow_server) as conn:
        cursor = conn.cursor()
        cursor.execute("SELECT 'hello from server'")
        assert cursor.fetchone() == ("hello from server",)
```

--------------------------------

### Loading Pandas DataFrames into Snowflake with write_pandas() in Python

Source: https://context7.com/tekumara/fakesnow/llms.txt

Provides examples of using the write_pandas function to efficiently load pandas DataFrames into Snowflake tables. It covers both loading into existing tables and automatically creating tables based on DataFrame structure.

```python
import fakesnow
import snowflake.connector
from snowflake.connector.pandas_tools import write_pandas
import pandas as pd

with fakesnow.patch():
    conn = snowflake.connector.connect(database="db1", schema="schema1")
    cursor = conn.cursor()

    # Create a DataFrame
    df = pd.DataFrame({
        "id": [1, 2, 3],
        "name": ["Alice", "Bob", "Charlie"],
        "score": [95, 87, 92]
    })

    # Create target table
    cursor.execute("CREATE TABLE students (id INT, name VARCHAR, score INT)")

    # Load DataFrame into table
    success, num_chunks, num_rows, output = write_pandas(
        conn,
        df,
        table_name="students"
    )

    print(f"Success: {success}, Rows loaded: {num_rows}")  # Output: Success: True, Rows loaded: 3

    # Verify data
    cursor.execute("SELECT * FROM students ORDER BY id")
    print(cursor.fetchall())  # Output: [(1, 'Alice', 95), (2, 'Bob', 87), (3, 'Charlie', 92)]

    # Auto-create table from DataFrame
    df2 = pd.DataFrame({"product": ["A", "B"], "quantity": [100, 200]})
    write_pandas(conn, df2, table_name="inventory", auto_create_table=True)

    cursor.execute("SELECT * FROM inventory")
    print(cursor.fetchall())  # Output: [('A', 100), ('B', 200)]
```

--------------------------------

### Retrieve Snowflake query results as pandas DataFrame

Source: https://context7.com/tekumara/fakesnow/llms.txt

Demonstrates how to use fetch_pandas_all() to convert Snowflake query results directly into a pandas DataFrame for data analysis. This requires the pandas library to be installed in the environment.

```python
import fakesnow
import snowflake.connector

with fakesnow.patch():
    conn = snowflake.connector.connect(database="db1", schema="schema1")
    cursor = conn.cursor()

    cursor.execute("CREATE TABLE metrics (date DATE, visits INT, conversions INT)")
    cursor.execute("""
        INSERT INTO metrics VALUES
        ('2024-01-01', 1000, 50),
        ('2024-01-02', 1200, 65),
        ('2024-01-03', 950, 42)
    """)

    cursor.execute("SELECT * FROM metrics ORDER BY date")

    df = cursor.fetch_pandas_all()
    print(df)
    print(f"Total visits: {df['VISITS'].sum()}")
    print(f"Avg conversion rate: {(df['CONVERSIONS'] / df['VISITS']).mean():.2%}")
```

--------------------------------

### Initialize DuckDB and SQL Magic in Python

Source: https://github.com/tekumara/fakesnow/blob/main/notebooks/examples/duckdb.ipynb

This snippet demonstrates how to initialize an in-memory DuckDB connection using Python and configure the IPython SQL magic extension for seamless SQL execution within a notebook environment. It sets up the SQL magic to use DuckDB and disables verbose feedback.

```python
import duckdb

duck_conn = duckdb.connect(database=":memory:")

%load_ext sql
%config SqlMagic.feedback = False
%config SqlMagic.displaycon = False
%config SqlMagic.autopandas = True
%sql duckdb:///:memory:
```

--------------------------------

### Configure Fakesnow Server with Custom Port and Database Persistence

Source: https://context7.com/tekumara/fakesnow/llms.txt

Shows how to configure the Fakesnow server to run on a specific port and utilize database persistence by providing a directory path. It also illustrates how to set up isolated, in-memory databases for each connection by using the ':isolated:' path.

```python
import fakesnow
import snowflake.connector
import tempfile

with tempfile.TemporaryDirectory() as db_dir:
    # Start server on specific port with persistent storage
    with fakesnow.server(
        port=18080,
        session_parameters={"FAKESNOW_DB_PATH": db_dir}
    ) as conn_kwargs:
        conn = snowflake.connector.connect(**conn_kwargs)
        cursor = conn.cursor()

        cursor.execute("CREATE DATABASE inventory")
        cursor.execute("CREATE TABLE inventory.public.items (sku VARCHAR, qty INT)")
        cursor.execute("INSERT INTO inventory.public.items VALUES ('ABC123', 50)")

        cursor.execute("SELECT * FROM inventory.public.items")
        print(cursor.fetchall())  # Output: [('ABC123', 50)]
        conn.close()

# For isolated connections (each connection gets its own database):
with fakesnow.server(
    session_parameters={"FAKESNOW_DB_PATH": ":isolated:"}
) as conn_kwargs:
    conn = snowflake.connector.connect(**conn_kwargs)
    # This connection has its own isolated in-memory database
    cursor = conn.cursor()
    cursor.execute("SELECT 'isolated connection' AS msg")
    print(cursor.fetchone())  # Output: ('isolated connection',)
```

--------------------------------

### Configure fakesnow Server Persistence

Source: https://github.com/tekumara/fakesnow/blob/main/README.md

Demonstrates how to configure the fakesnow server for database persistence or isolation using session parameters.

```python
# Persist to directory
with fakesnow.server(session_parameters={"FAKESNOW_DB_PATH": "databases/"}):
    ...

# Isolated in-memory database
with fakesnow.server(session_parameters={"FAKESNOW_DB_PATH": ":isolated:"}):
    ...
```

--------------------------------

### Bind parameters in SQL queries

Source: https://context7.com/tekumara/fakesnow/llms.txt

Illustrates how to use qmark and pyformat styles for secure parameter binding in Snowflake queries. It also demonstrates using the IDENTIFIER keyword for dynamic table names.

```python
import fakesnow
import snowflake.connector

with fakesnow.patch():
    conn = snowflake.connector.connect(database="db1", schema="schema1")
    cursor = conn.cursor()

    cursor.execute("CREATE TABLE users (id INT, name VARCHAR, age INT)")
    cursor.execute("INSERT INTO users VALUES (?, ?, ?)", (1, "Alice", 30))

    cursor.execute(
        "SELECT * FROM users WHERE name = %(name)s AND age > %(min_age)s",
        {"name": "Alice", "min_age": 25}
    )
    print(cursor.fetchall())

    table_name = "users"
    cursor.execute("SELECT * FROM IDENTIFIER(?)", (table_name,))
    print(cursor.fetchall())
```

--------------------------------

### Mock Snowflake Connection with Fakesnow (Python)

Source: https://github.com/tekumara/fakesnow/blob/main/notebooks/examples/fakesnow.ipynb

This snippet demonstrates how to use Fakesnow to mock a Snowflake connection using the snowflake-connector library. It shows setting up the mock environment, establishing a connection, and executing a simple SELECT query, returning a mocked result.

```python
import snowflake.connector

import fakesnow

p = fakesnow.patch()
p.__enter__()
conn = snowflake.connector.connect(database='db1', schema='schema1')
cur = conn.cursor()
print(cur.execute("SELECT 'Hello fake world!'").fetchone())
```

--------------------------------

### Load Data from S3 using COPY INTO

Source: https://context7.com/tekumara/fakesnow/llms.txt

Demonstrates loading data into Snowflake tables from S3 buckets using the COPY INTO command, including configuring S3 secrets for custom endpoints.

```python
import fakesnow
import snowflake.connector

with fakesnow.patch():
    conn = snowflake.connector.connect(database="db1", schema="schema1")
    cursor = conn.cursor()
    cursor.execute("CREATE SECRET aws_secret (TYPE S3, KEY_ID 'test', SECRET 'test', ENDPOINT 'localhost:5000', USE_SSL false)")
    cursor.execute("COPY INTO sales FROM 's3://my-bucket/data/sales.csv' FILE_FORMAT = (TYPE = CSV SKIP_HEADER = 1)")
    cursor.execute("COPY INTO sales FROM 's3://my-bucket/data/sales.parquet' FILE_FORMAT = (TYPE = PARQUET)")
```

--------------------------------

### Inspect Data Types and Simple Queries

Source: https://github.com/tekumara/fakesnow/blob/main/notebooks/examples/snowflake.ipynb

Demonstrates using system functions to inspect data types and basic timestamp conversion queries.

```python
%sql SELECT system$typeof(to_date(to_timestamp(0)))
cur.execute("SELECT to_timestamp('2013-04-05 01:02:03')").fetchall()
```

--------------------------------

### In-Process Patching with fakesnow.patch()

Source: https://context7.com/tekumara/fakesnow/llms.txt

Demonstrates basic in-process patching using the `fakesnow.patch()` context manager. This allows connecting to a fake Snowflake instance without real credentials and executing standard SQL queries.

```python
import fakesnow
import snowflake.connector

# Basic usage with context manager
with fakesnow.patch():
    # Connect without real credentials - any values work
    conn = snowflake.connector.connect(
        user="test_user",
        password="test_pass",
        account="test_account",
        database="mydb",
        schema="myschema"
    )

    # Execute queries as you would with real Snowflake
    cursor = conn.cursor()
    cursor.execute("CREATE TABLE users (id INT, name VARCHAR, email VARCHAR)")
    cursor.execute("INSERT INTO users VALUES (1, 'Alice', 'alice@example.com')")
    cursor.execute("INSERT INTO users VALUES (2, 'Bob', 'bob@example.com')")

    cursor.execute("SELECT * FROM users WHERE id = 1")
    result = cursor.fetchone()
    print(result)  # Output: (1, 'Alice', 'alice@example.com')

    cursor.execute("SELECT COUNT(*) FROM users")
    print(cursor.fetchone())  # Output: (2,)

    conn.close()
```

--------------------------------

### Manage Transactions with Commit and Rollback

Source: https://context7.com/tekumara/fakesnow/llms.txt

Shows how to control transaction boundaries using BEGIN, COMMIT, and ROLLBACK commands within a fakesnow-patched Snowflake connection.

```python
import fakesnow
import snowflake.connector

with fakesnow.patch():
    conn = snowflake.connector.connect(database="db1", schema="schema1", autocommit=False)
    cursor = conn.cursor()
    cursor.execute("BEGIN")
    cursor.execute("UPDATE accounts SET balance = balance - 100 WHERE id = 1")
    conn.commit()
    cursor.execute("BEGIN")
    cursor.execute("UPDATE accounts SET balance = 0 WHERE id = 1")
    conn.rollback()
```

--------------------------------

### Database Persistence with db_path

Source: https://context7.com/tekumara/fakesnow/llms.txt

Demonstrates how to persist fake Snowflake databases to disk using the `db_path` argument in `fakesnow.patch()`. This allows data to survive between different patching sessions.

```python
import fakesnow
import snowflake.connector
import tempfile
import os

# Create a temporary directory for database files
with tempfile.TemporaryDirectory() as db_dir:
    # First session - create and populate data
    with fakesnow.patch(db_path=db_dir):
        conn = snowflake.connector.connect(database="analytics", schema="public")
        cursor = conn.cursor()

        cursor.execute("CREATE TABLE metrics (date DATE, value FLOAT)")
        cursor.execute("INSERT INTO metrics VALUES ('2024-01-01', 100.5)")
        cursor.execute("INSERT INTO metrics VALUES ('2024-01-02', 150.3)")
        conn.close()

    # Second session - data persists
    with fakesnow.patch(db_path=db_dir):
        conn = snowflake.connector.connect(database="analytics", schema="public")
        cursor = conn.cursor()

        cursor.execute("SELECT * FROM metrics ORDER BY date")
        results = cursor.fetchall()
        print(results)
        # Output: [(datetime.date(2024, 1, 1), 100.5), (datetime.date(2024, 1, 2), 150.3)]

        # Verify database file exists
        print(os.path.exists(os.path.join(db_dir, "ANALYTICS.db")))  # Output: True
```

--------------------------------

### Filter Snowflake Commands with Regex

Source: https://context7.com/tekumara/fakesnow/llms.txt

Demonstrates how to use regex patterns to silently skip specific SQL commands when using fakesnow. This is useful for ignoring unsupported or irrelevant DDL statements during testing.

```python
import fakesnow
import snowflake.connector

nop_patterns = [
    r"ALTER WAREHOUSE.*",
    r"CREATE STREAM.*",
    r"CREATE TASK.*",
]

with fakesnow.patch(nop_regexes=nop_patterns):
    conn = snowflake.connector.connect(database="db1", schema="schema1")
    cursor = conn.cursor()
    cursor.execute("ALTER WAREHOUSE my_warehouse RESUME")
    cursor.execute("CREATE STREAM my_stream ON TABLE my_table")
    cursor.execute("CREATE TABLE actual_table (id INT)")
```

--------------------------------

### Query Information Schema with Fakesnow Python

Source: https://context7.com/tekumara/fakesnow/llms.txt

This snippet shows how to use fakesnow to patch the Snowflake Connector and query the information schema for tables and columns. It also demonstrates executing SHOW commands. The code requires the `fakesnow` and `snowflake-connector-python` libraries.

```python
import fakesnow
import snowflake.connector

with fakesnow.patch():
    conn = snowflake.connector.connect(database="db1", schema="schema1")
    cursor = conn.cursor()

    cursor.execute("CREATE TABLE users (id INT, name VARCHAR(50), email VARCHAR(100))")
    cursor.execute("CREATE TABLE orders (id INT, user_id INT, total DECIMAL(10,2))")
    cursor.execute("CREATE VIEW user_orders AS SELECT u.name, o.total FROM users u JOIN orders o ON u.id = o.user_id")

    # List tables
    cursor.execute("""
        SELECT table_name, table_type
        FROM information_schema.tables
        WHERE table_schema = 'SCHEMA1'
        ORDER BY table_name
    """)
    print(cursor.fetchall())
    # Output: [('ORDERS', 'BASE TABLE'), ('USERS', 'BASE TABLE'), ('USER_ORDERS', 'VIEW')]

    # Get column information
    cursor.execute("""
        SELECT column_name, data_type, character_maximum_length
        FROM information_schema.columns
        WHERE table_name = 'USERS'
        ORDER BY ordinal_position
    """)
    print(cursor.fetchall())
    # Output: [('ID', 'NUMBER', None), ('NAME', 'TEXT', 50), ('EMAIL', 'TEXT', 100)]

    # SHOW commands
    cursor.execute("SHOW TABLES")
    cursor.execute("SHOW SCHEMAS")
    cursor.execute("SHOW DATABASES")
```

--------------------------------

### Execute SQL Queries with Snowflake Connector Cursor

Source: https://context7.com/tekumara/fakesnow/llms.txt

Illustrates how to use the `cursor.execute()` method within a Fakesnow patched environment to run various SQL statements. This includes DDL operations, parameterized inserts, batch inserts, and different fetching methods (`fetchone`, `fetchall`, `fetchmany`), as well as accessing `rowcount`.

```python
import fakesnow
import snowflake.connector

with fakesnow.patch():
    conn = snowflake.connector.connect(database="db1", schema="schema1")
    cursor = conn.cursor()

    # DDL operations
    cursor.execute("CREATE TABLE products (id INT, name VARCHAR, price DECIMAL(10,2))")
    print(cursor.fetchone())  # Output: ('Table PRODUCTS successfully created.',)

    # Insert with parameters (qmark style)
    cursor.execute(
        "INSERT INTO products VALUES (?, ?, ?)",
        (1, "Widget", 29.99)
    )

    # Insert multiple rows
    cursor.executemany(
        "INSERT INTO products VALUES (?, ?, ?)",
        [(2, "Gadget", 49.99), (3, "Gizmo", 19.99)]
    )

    # Query with fetchall
    cursor.execute("SELECT * FROM products ORDER BY id")
    rows = cursor.fetchall()
    print(rows)
    # Output: [(1, 'Widget', Decimal('29.99')), (2, 'Gadget', Decimal('49.99')), (3, 'Gizmo', Decimal('19.99'))]

    # Query with fetchone
    cursor.execute("SELECT * FROM products WHERE id = 2")
    print(cursor.fetchone())  # Output: (2, 'Gadget', Decimal('49.99'))

    # Query with fetchmany
    cursor.execute("SELECT * FROM products ORDER BY id")
    print(cursor.fetchall())
    # Output: [(1, 'Widget', Decimal('29.99')), (2, 'Gadget', Decimal('49.99'))]

    # Access rowcount
    cursor.execute("UPDATE products SET price = 24.99 WHERE id = 1")
    print(cursor.rowcount)  # Output: 1
```

--------------------------------

### Retrieve and Cast Current Timestamps

Source: https://github.com/tekumara/fakesnow/blob/main/notebooks/examples/timezones.md

Demonstrates how to fetch the current system time and cast it to different precision levels or timezone-agnostic formats. Note that behavior varies significantly regarding implicit timezone handling and casting limitations between database engines.

```sql
select current_timestamp;
select current_timestamp::TIMESTAMP;
select current_timestamp::TIMESTAMP::TIMESTAMP(9);
select current_timestamp AT TIME ZONE 'UTC';
select (current_timestamp AT TIME ZONE 'UTC')::TIMESTAMP(9);
select current_timestamp::TIMESTAMP(9);
```

```sql
select current_timestamp;
select current_timestamp::TIMESTAMP;
```

--------------------------------

### Running fakesnow as an HTTP Server

Source: https://context7.com/tekumara/fakesnow/llms.txt

Introduces the concept of running fakesnow as a standalone HTTP server. This mode is suitable for non-Python clients or scenarios involving subprocesses.

```python
import fakesnow
import snowflake.connector
```

--------------------------------

### Configure Snowflake CLI for Fakesnow

Source: https://github.com/tekumara/fakesnow/blob/main/README.md

Defines the connection settings in config.toml to point the Snowflake CLI to a local Fakesnow instance. It is critical to set the protocol to http for compatibility.

```toml
[connections.fakesnow]
host = "localhost"
port = <port number from server startup>
account = "fakesnow"
user = "fake"
password = "snow"
protocol = "http"
```

--------------------------------

### Execute Timestamp Queries

Source: https://github.com/tekumara/fakesnow/blob/main/notebooks/examples/snowflake.ipynb

Demonstrates how to cast and select timestamp_tz and timestamp data types using both the Python connector cursor and SQL Magic.

```python
cur.execute(
    "select '2020-12-31 00:01:02.345678 +0000'::timestamp_tz::timestamp, '2020-12-31 00:01:02.345678+00:00'::timestamp_tz, '2020-12-31 00:01:02.345678+10:00'::timestamp_tz, '2020-12-31 00:01:02.345678+10:00'::timestamp_tz::timestamp"
).fetchall()

%sql select '2020-12-31 00:01:02.345678+00:00'::timestamp_tz::timestamp, '2020-12-31 00:01:02.345678+00:00'::timestamp_tz, '2020-12-31 00:01:02.345678+10:00'::timestamp_tz, '2020-12-31 00:01:02.345678 +1000'::timestamp_tz
```

--------------------------------

### Connect to Snowflake using Python Connector

Source: https://github.com/tekumara/fakesnow/blob/main/notebooks/examples/snowflake.ipynb

Establishes a connection to a Snowflake database using the snowflake-connector-python library with a dictionary of connection parameters.

```python
import snowflake.connector

conn_info = dict(
    user = ...,
    role = ...,
    account = ...,
    authenticator=...,
    database=...,
    schema=...,
)
conn = snowflake.connector.connect(**conn_info)
cur = conn.cursor()
```

--------------------------------

### Manual Patching for Imports

Source: https://github.com/tekumara/fakesnow/blob/main/README.md

Shows how to patch specific modules when using the 'from ... import' syntax, which is not automatically handled by the default patcher.

```python
with fakesnow.patch("mymodule.write_pandas"):
    ...
```

--------------------------------

### Using DictCursor for Dictionary Results in Python

Source: https://context7.com/tekumara/fakesnow/llms.txt

Demonstrates how to use fakesnow's DictCursor to fetch query results as Python dictionaries, allowing access to columns by name. This simplifies data manipulation compared to traditional tuple-based results.

```python
import fakesnow
import snowflake.connector
from snowflake.connector.cursor import DictCursor

with fakesnow.patch():
    conn = snowflake.connector.connect(database="db1", schema="schema1")

    # Create cursor with DictCursor class
    cursor = conn.cursor(DictCursor)

    cursor.execute("CREATE TABLE employees (id INT, name VARCHAR, department VARCHAR)")
    cursor.execute("INSERT INTO employees VALUES (1, 'John', 'Engineering')")
    cursor.execute("INSERT INTO employees VALUES (2, 'Jane', 'Marketing')")

    cursor.execute("SELECT * FROM employees ORDER BY id")
    results = cursor.fetchall()
    print(results)
    # Output: [{'ID': 1, 'NAME': 'John', 'DEPARTMENT': 'Engineering'},
    #          {'ID': 2, 'NAME': 'Jane', 'DEPARTMENT': 'Marketing'}]

    # Access by column name
    for row in results:
        print(f"{row['NAME']} works in {row['DEPARTMENT']}")
```

--------------------------------

### Configure SQL Magic for Jupyter

Source: https://github.com/tekumara/fakesnow/blob/main/notebooks/examples/snowflake.ipynb

Sets up the SQL Magic extension in Jupyter to execute SQL queries directly against a Snowflake engine created via SQLAlchemy.

```python
from snowflake.sqlalchemy import URL
from sqlalchemy import create_engine

url = URL(**conn_info)
engine = create_engine(url)

%reload_ext sql

%config SqlMagic.feedback = False
%config SqlMagic.displaycon = False
%config SqlMagic.autopandas = True
%sql engine
%sql select current_warehouse(), current_database(), current_schema()
```

--------------------------------

### Cursor.execute()

Source: https://context7.com/tekumara/fakesnow/llms.txt

Executes SQL statements against the mocked Snowflake environment using the standard cursor interface.

```APIDOC
## Cursor.execute()

### Description
Executes SQL queries (DDL, DML, DQL) against the mocked database. Supports standard Snowflake SQL syntax and parameter binding.

### Method
SQL Execution

### Parameters
- **sql** (str) - Required - The SQL statement to execute.
- **params** (tuple) - Optional - Parameters for qmark style binding.

### Request Example
cursor.execute("INSERT INTO products VALUES (?, ?, ?)", (1, "Widget", 29.99))

### Response
#### Success Response
- **result** (tuple/list) - Returns query results via fetchone(), fetchall(), or fetchmany().
```

--------------------------------

### Test with Fakesnow Server Fixture

Source: https://github.com/tekumara/fakesnow/blob/main/README.md

Utilizes the fakesnow_server fixture to spin up a server instance and connect using the standard Snowflake Python connector for integration testing.

```python
import snowflake.connector

def test_with_server(fakesnow_server: dict):
    with snowflake.connector.connect(**fakesnow_server) as conn:
        conn.cursor().execute("SELECT 1")
        assert conn.cursor().fetchone() == (1,)
```

--------------------------------

### In-process Patching with fakesnow

Source: https://github.com/tekumara/fakesnow/blob/main/README.md

Demonstrates how to use fakesnow.patch to intercept Snowflake connector calls within a Python script. This ensures database operations are routed to the local mock instead of a real Snowflake instance.

```python
import fakesnow
import snowflake.connector

with fakesnow.patch():
    conn = snowflake.connector.connect()
    print(conn.cursor().execute("SELECT 'Hello fake world!'").fetchone())
```

--------------------------------

### Patching Modules with Fakesnow

Source: https://github.com/tekumara/fakesnow/blob/main/README.md

Demonstrates how to use the fakesnow.patch context manager within a pytest fixture to intercept and mock specific module functions during tests.

```python
import fakesnow
import pytest

@pytest.fixture(scope="session", autouse=True)
def _fakesnow_session() -> Iterator[None]:
    with fakesnow.patch("mymodule.write_pandas"):
        yield
```

--------------------------------

### Process large result sets in batches

Source: https://context7.com/tekumara/fakesnow/llms.txt

Shows how to use get_result_batches() to retrieve and process large datasets in memory-efficient chunks. Each batch can be iterated row-by-row or converted to a pandas DataFrame.

```python
import fakesnow
import snowflake.connector

with fakesnow.patch():
    conn = snowflake.connector.connect(database="db1", schema="schema1")
    cursor = conn.cursor()

    cursor.execute("CREATE TABLE large_table (id INT, value VARCHAR)")
    cursor.executemany(
        "INSERT INTO large_table VALUES (?, ?)",
        [(i, f"value_{i}") for i in range(100)]
    )

    cursor.execute("SELECT * FROM large_table ORDER BY id")

    batches = cursor.get_result_batches()
    total_rows = 0
    for batch in batches:
        for row in batch.create_iter():
            total_rows += 1
        df = batch.to_pandas()
        print(f"Batch size: {batch.rowcount}")
    print(f"Total rows processed: {total_rows}")
```

--------------------------------

### Accessing Query Result Metadata with cursor.description in Python

Source: https://context7.com/tekumara/fakesnow/llms.txt

Illustrates how to use the cursor.description attribute to retrieve metadata about the columns returned by a SQL query. This includes column names, data types, precision, and scale, which is useful for understanding query results.

```python
import fakesnow
import snowflake.connector

with fakesnow.patch():
    conn = snowflake.connector.connect(database="db1", schema="schema1")
    cursor = conn.cursor()

    cursor.execute("""
        CREATE TABLE products (
            id INT,
            name VARCHAR(100),
            price DECIMAL(10,2),
            in_stock BOOLEAN,
            created_at TIMESTAMP
        )
    """)

    cursor.execute("SELECT * FROM products")

    # Get column metadata
    for col in cursor.description:
        print(f"Name: {col.name}, Type: {col.type_code}, Precision: {col.precision}, Scale: {col.scale}")
    # Output:
    # Name: ID, Type: 0, Precision: 38, Scale: 0
    # Name: NAME, Type: 2, Precision: None, Scale: None
    # Name: PRICE, Type: 0, Precision: 10, Scale: 2
    # Name: IN_STOCK, Type: 13, Precision: None, Scale: None
    # Name: CREATED_AT, Type: 8, Precision: 0, Scale: 9
```

--------------------------------

### Python: Transform Snowflake SQL to DuckDB with sqlglot

Source: https://github.com/tekumara/fakesnow/blob/main/notebooks/examples/sqlglot.ipynb

This snippet uses the sqlglot library in Python to parse a Snowflake SQL query containing `parse_json` and `lateral flatten` and then transforms it into DuckDB compatible SQL syntax. It highlights the library's capability to handle dialect-specific functions and structures.

```python
import sqlglot
from sqlglot import exp

import fakesnow.transforms as transforms

sqlglot.parse_one(
            """
            select t.id, flat.value:fruit from
            (
                select 1, parse_json('[{"fruit":"banana"}]')
                union
                select 2, parse_json('[{"fruit":"coconut"}, {"fruit":"durian"}]')
            ) as t(id, fruits), lateral flatten(input => t.fruits)
            """, read="snowflake").sql(dialect="duckdb")
```

--------------------------------

### Execute JSON Array Query in DuckDB Python

Source: https://github.com/tekumara/fakesnow/blob/main/notebooks/examples/duckdb.ipynb

This Python code snippet shows how to execute a SQL query against a DuckDB connection to create a JSON array from a list of strings. The result is fetched and returned as a list of tuples.

```python
duck_conn.execute("select json(['A', 'B'])").fetchall()
```

--------------------------------

### Executing Multiple SQL Statements with execute_string() in Python

Source: https://context7.com/tekumara/fakesnow/llms.txt

Shows how to execute a string containing multiple SQL statements using conn.execute_string(). This method returns an iterable of cursors, one for each statement, enabling batch operations and retrieval of results from individual statements.

```python
import fakesnow
import snowflake.connector

with fakesnow.patch():
    conn = snowflake.connector.connect(database="db1", schema="schema1")

    # Execute multiple statements at once
    cursors = conn.execute_string("""
        CREATE TABLE orders (id INT, customer_id INT, total DECIMAL(10,2));
        INSERT INTO orders VALUES (1, 100, 250.00);
        INSERT INTO orders VALUES (2, 101, 175.50);
        INSERT INTO orders VALUES (3, 100, 99.99);
        SELECT customer_id, SUM(total) as total_spent
        FROM orders
        GROUP BY customer_id
        ORDER BY total_spent DESC;
    """)

    # Get results from the last cursor (the SELECT)
    last_cursor = list(cursors)[-1]
    results = last_cursor.fetchall()
    print(results)  # Output: [(100, Decimal('349.99')), (101, Decimal('175.50'))]

    # Check rowcount from insert
    insert_cursor = list(cursors)[2]
    print(insert_cursor.rowcount)  # Output: 1
```

--------------------------------

### Patching Extra Targets for 'from ... import' Statements

Source: https://context7.com/tekumara/fakesnow/llms.txt

Shows how to patch specific targets when code uses `from ... import` syntax for Snowflake connector modules. This ensures that fakesnow intercepts calls to these explicitly imported functions.

```python
import fakesnow
import snowflake.connector

# If your module uses: from snowflake.connector.pandas_tools import write_pandas
# Patch it explicitly
with fakesnow.patch("mymodule.write_pandas"):
    conn = snowflake.connector.connect(database="db1", schema="public")
    cursor = conn.cursor()

    # Create a table
    cursor.execute("CREATE TABLE events (event_id INT, event_name VARCHAR)")
    cursor.execute("INSERT INTO events VALUES (1, 'Login'), (2, 'Purchase')")

    # Your module's write_pandas calls will now use the fake
    cursor.execute("SELECT * FROM events")
    print(cursor.fetchall())  # Output: [(1, 'Login'), (2, 'Purchase')]
```

--------------------------------

### Perform Semi-Structured Data Operations

Source: https://context7.com/tekumara/fakesnow/llms.txt

Covers working with VARIANT data types, including parsing JSON, accessing nested fields, flattening arrays, and using aggregation functions like OBJECT_CONSTRUCT and ARRAY_AGG.

```python
import fakesnow
import snowflake.connector

with fakesnow.patch():
    conn = snowflake.connector.connect(database="db1", schema="schema1")
    cursor = conn.cursor()
    cursor.execute("CREATE TABLE events (id INT, data VARIANT)")
    cursor.execute("INSERT INTO events SELECT 1, PARSE_JSON('{\"user\": \"alice\", \"action\": \"login\", \"metadata\": {\"ip\": \"192.168.1.1\"}}')")
    cursor.execute("SELECT id, data:user::VARCHAR as user, data:action::VARCHAR as action FROM events")
    cursor.execute("SELECT data:metadata.ip::VARCHAR as ip FROM events WHERE id = 1")
    cursor.execute("SELECT e.id, f.value::VARCHAR as item FROM events e, LATERAL FLATTEN(input => e.data:items) f WHERE e.id = 2")
```

--------------------------------

### Write Pandas DataFrame to Snowflake

Source: https://github.com/tekumara/fakesnow/blob/main/notebooks/examples/snowflake.ipynb

Uses the snowflake-connector-pandas-tools to write a pandas DataFrame containing UTC timestamps into a Snowflake table.

```python
import datetime
import pandas as pd
import pytz
import snowflake.connector.pandas_tools

cur.execute("create or replace table example (UPDATE_AT_NTZ timestamp_ntz(9))")

now_utc = datetime.datetime.now(pytz.utc)
df = pd.DataFrame([(now_utc,)], columns=["UPDATE_AT_NTZ"])
snowflake.connector.pandas_tools.write_pandas(conn, df, "EXAMPLE")

cur.execute("select * from example")
assert cur.fetchall() == [(now_utc.replace(tzinfo=None),)]
```

=== COMPLETE CONTENT === This response contains all available snippets from this library. No additional content exists. Do not make further requests.