tekumara

tekumara /fakesnow

Run, mock and test fake Snowflake databases locally.

143
14
GitHub
Public
Apache License 2.0
759 KB

Repository Statistics

Key metrics and engagement data

143
Stars
14
Forks
14
Open Issues
0
Releases
1.10
Engagement Rate
Default branch: main

Timeline

Repository has been active for 2 years, 5 months

Repository Created

Last Commit
Recently active

README.md

fakesnow ❄️

ci release PyPI PyPI - Downloads

Run, mock and test fake Snowflake databases locally.

Install

1pip install fakesnow

Or to install with the server:

1pip install fakesnow[server]

Usage

fakesnow offers two main approaches for faking Snowflake: in-process patching of the Snowflake Connector for Python or a standalone HTTP server.

Patching only applies to the current Python process. If a subprocess is spawned it won't be patched. For subprocesses, or for non-Python clients, use the server instead.

In-process patching

To run script.py with patching:

shell
1fakesnow script.py

Or a module, eg: pytest

shell
1fakesnow -m pytest

fakesnow executes fakesnow.patch before running the script or module.

Use fakesnow.patch in your code

Alternatively, use fakesnow.patch in your code:

python
1import fakesnow
2import snowflake.connector
3
4with fakesnow.patch():
5 conn = snowflake.connector.connect()
6
7 print(conn.cursor().execute("SELECT 'Hello fake world!'").fetchone())

What gets patched

The following standard imports are automatically patched:

  • import snowflake.connector.connect
  • import snowflake.connector.pandas_tools.write_pandas

Handling "from ... import" Statements

To patch modules that use the from ... import syntax, you need to manually specify them, eg: if mymodule.py contains:

python
1from snowflake.connector.pandas_tools import write_pandas

Then patch it using:

python
1with fakesnow.patch("mymodule.write_pandas"):
2 ...

Database Persistence

By default, databases are in-memory and will be lost when the process ends. To persist databases between processes, specify a databases path:

python
1with fakesnow.patch(db_path="databases/"):
2 ...

Run fakesnow as a server

For scenarios where patching won't work (like subprocesses or non-Python clients), you can run fakesnow as an HTTP server.

From the command line

Using uv:

1uvx 'fakesnow[server]' -s

Or from within a virtualenv that has fakesnow[server] installed:

1fakesnow -s

By default the server listens on a random available port. Use -p to specify a port.

Within your python program

python
1import fakesnow
2import snowflake.connector
3
4# Start the fakesnow server in a context manager
5# This yields connection kwargs (host, port, etc.)
6with fakesnow.server() as conn_kwargs:
7 # Connect to the fakesnow server using the yielded kwargs
8 with snowflake.connector.connect(**conn_kwargs) as conn:
9 print(conn.cursor().execute("SELECT 'Hello fake server!'").fetchone())
10
11 # The server is automatically stopped when exiting the context manager

This starts an HTTP server in its own thread listening for requests on a random available port.

Server Configuration Options

To specify a port for the server:

python
1with fakesnow.server(port=12345) as conn_kwargs:
2 ...

By default, the server uses a single in-memory database for its lifetime. To configure database persistence or isolation:

python
1# Databases will be saved to the "databases/" directory
2with fakesnow.server(session_parameters={"FAKESNOW_DB_PATH": "databases/"}):
3 ...
4
5# Each connection gets its own isolated in-memory database
6with fakesnow.server(session_parameters={"FAKESNOW_DB_PATH": ":isolated:"}):
7 ...

Connecting from non-Python clients

The server is available via HTTP and accepts any username/password/account combination, eg:

1user: fake
2password: snow
3account: fakesnow
4host: localhost
5port: <port number from server startup>
6protocol: http

Additional parameters that may be helpful:

  • Session parameter CLIENT_OUT_OF_BAND_TELEMETRY_ENABLED set to false
  • Network timeout set to 1 second (since retries aren't needed in testing)

For example, with the Snowflake CLI:

1snowsql -a fakesnow -u fake -p snow -h localhost -P <port> --protocol http

pytest fixtures

fakesnow provides fixtures for easier test integration. Add them in conftest.py:

python
1pytest_plugins = "fakesnow.fixtures"

To autouse the fixture you can wrap it like this in conftest.py:

python
1from typing import Iterator
2
3import pytest
4
5pytest_plugins = "fakesnow.fixtures"
6
7@pytest.fixture(scope="session", autouse=True)
8def setup(_fakesnow_session: None) -> Iterator[None]:
9 # the standard imports are now patched
10 # Add any additional setup here
11 yield
12 # Add any teardown here

For code that uses from ... import statements:

python
1from typing import Iterator
2
3import fakesnow
4import pytest
5
6pytest_plugins = "fakesnow.fixtures"
7
8@pytest.fixture(scope="session", autouse=True)
9def _fakesnow_session() -> Iterator[None]:
10 with fakesnow.patch("mymodule.write_pandas"):
11 yield

server fixture

To start a fakesnow server instance, enable the plugin in conftest.py:

python
1pytest_plugins = "fakesnow.fixtures"

And then use the fakesnow_server session fixture like this:

python
1import snowflake.connector
2
3def test_with_server(fakesnow_server: dict):
4 # fakesnow_server contains connection kwargs (host, port, etc.)
5 with snowflake.connector.connect(**fakesnow_server) as conn:
6 conn.cursor().execute("SELECT 1")
7 assert conn.cursor().fetchone() == (1,)

Implementation coverage

Fully supported:

Partially supported:

  • Date functions
  • Regular expression functions
  • Semi-structured data operations
  • Tags
  • User management
  • Stages and PUT
  • COPY INTO from S3 sources and stages, see COPY INTO

Not yet implemented:

For more detail see the test suite.

Caveats

  • Row ordering is non-deterministic and may differ from Snowflake unless you fully specify the ORDER BY clause.
  • fakesnow supports a more liberal SQL dialect than actual Snowflake. This means some queries that work with fakesnow might not work with a real Snowflake instance.

COPY INTO

COPY INTO can be used from S3 sources and stages. By default the standard AWS credential chain will be used. If you are getting an HTTP 403 or need to provide alternative S3 credentials you can use the duckdb CREATE SECRET statement. For an example of creating a secret to use a moto S3 endpoint see s3_client in conftest.py

Contributing

See CONTRIBUTING.md for instructions on getting started with development and contributing to this project.