Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions README.md
Original file line number Diff line number Diff line change
Expand Up @@ -489,6 +489,8 @@ metadata.create_all(engine)

This also applies when using pandas `to_sql()` with timezone-aware datetime columns, which infers `DateTime(timezone=True)` automatically (see [#199](https://github.com/snowflakedb/snowflake-sqlalchemy/issues/199)).

**Note on `Time` and timezones:** SQLAlchemy's `Time` type accepts a `timezone` parameter, but [Snowflake's TIME data type does not support time zones](https://docs.snowflake.com/en/sql-reference/data-types-datetime#time). Using `Time(timezone=True)` will compile to plain `TIME` and the `timezone` flag will have no effect. If you need to store time data with time-zone information, use a timestamp type such as `TIMESTAMP_TZ` or `DateTime(timezone=True)` instead.

### Cache Column Metadata

SQLAlchemy provides [the runtime inspection API](http://docs.sqlalchemy.org/en/latest/core/inspection.html) to get the runtime information about the various objects. One of the common use case is get all tables and their column metadata in a schema in order to construct a schema catalog. For example, [alembic](http://alembic.zzzcomputing.com/) on top of SQLAlchemy manages database schema migrations. A pseudo code flow is as follows:
Expand Down
10 changes: 10 additions & 0 deletions src/snowflake/sqlalchemy/custom_types.py
Original file line number Diff line number Diff line change
Expand Up @@ -265,6 +265,16 @@ def process(value):


class _CUSTOM_Time(SnowflakeType, sqltypes.Time):
"""Internal Time type for the Snowflake dialect.

SQLAlchemy's ``Time(timezone=True)`` has no effect in this dialect because
Snowflake's TIME data type does not support time zones
(https://docs.snowflake.com/en/sql-reference/data-types-datetime#time).
The column will always be compiled to plain ``TIME`` regardless of the
``timezone`` flag. To store timestamps with time-zone information use
:class:`TIMESTAMP_TZ` or ``DateTime(timezone=True)`` instead.
"""

def literal_processor(self, dialect):
def process(value):
if value is not None:
Expand Down
33 changes: 31 additions & 2 deletions tests/test_timestamp.py
Original file line number Diff line number Diff line change
Expand Up @@ -7,13 +7,13 @@
import pandas as pd
import pytest
import pytz
from sqlalchemy import Column, DateTime, Integer, MetaData, Table, inspect, text
from sqlalchemy import Column, DateTime, Integer, MetaData, Table, Time, inspect, text
from sqlalchemy.schema import CreateTable
from sqlalchemy.sql import select
from sqlalchemy.types import TIMESTAMP

from snowflake.sqlalchemy import TIMESTAMP_LTZ, TIMESTAMP_NTZ, TIMESTAMP_TZ, snowdialect
from snowflake.sqlalchemy.custom_types import _CUSTOM_DateTime
from snowflake.sqlalchemy.custom_types import _CUSTOM_DateTime, _CUSTOM_Time
from tests.util import compile_type, normalize_ddl, random_string

PST_TZ = "America/Los_Angeles"
Expand Down Expand Up @@ -103,6 +103,35 @@ def test_create_table_explicit_snowflake_types_ddl(self):
assert "ltz TIMESTAMP_LTZ" in ddl


class TestUnitCustomTime:
"""Unit tests for _CUSTOM_Time literal processor and timezone-ignored compilation."""

def test_time_with_timezone_compiles_to_plain_time(self):
assert compile_type(Time(timezone=True)) == "TIME"

def test_time_without_timezone_compiles_to_time(self):
assert compile_type(Time(timezone=False)) == "TIME"

def test_custom_time_literal_processor_formats_microseconds(self):
from datetime import time

custom_time = _CUSTOM_Time()
processor = custom_time.literal_processor(dialect=snowdialect.dialect())
assert processor(time(14, 30, 59, 123456)) == "'14:30:59.123456'"

def test_custom_time_literal_processor_zero_microseconds(self):
from datetime import time

custom_time = _CUSTOM_Time()
processor = custom_time.literal_processor(dialect=snowdialect.dialect())
assert processor(time(8, 0, 0)) == "'08:00:00.000000'"

def test_custom_time_literal_processor_returns_none_for_none(self):
custom_time = _CUSTOM_Time()
processor = custom_time.literal_processor(dialect=snowdialect.dialect())
assert processor(None) is None


class TestIntegrationDatetimeAndTimestampWithTimezone:
"""Integration tests for issue #199 against a real Snowflake account."""

Expand Down
Loading