Added infrastructure to support PR #20 (TimestampTZ repr)

Author: erezsh

sqeleton/abcs/database_types.py

@@ -5,7 +5,7 @@
 
 from runtype import dataclass
 
-from ..utils import ArithAlphanumeric, ArithUUID, Self
+from ..utils import ArithAlphanumeric, ArithUUID, Self, Unknown
 
 
 DbPath = Tuple[str, ...]
@@ -20,7 +20,7 @@ class ColType:
 @dataclass
 class PrecisionType(ColType):
     precision: int
-    rounds: bool
+    rounds: Union[bool, Unknown] = Unknown
 
 
 class Boolean(ColType):

sqeleton/databases/base.py

@@ -30,6 +30,7 @@
     String_VaryingAlphanum,
     TemporalType,
     UnknownColType,
+    TimestampTZ,
     Text,
     DbTime,
     DbPath,
@@ -202,6 +203,8 @@ def constant_values(self, rows) -> str:
     def type_repr(self, t) -> str:
         if isinstance(t, str):
             return t
+        elif isinstance(t, TimestampTZ):
+            return f"TIMESTAMP({min(t.precision, DEFAULT_DATETIME_PRECISION)})"
         return {
             int: "INT",
             str: "VARCHAR",

sqeleton/databases/postgresql.py

@@ -97,6 +97,11 @@ def set_timezone_to_utc(self) -> str:
     def current_timestamp(self) -> str:
         return "current_timestamp"
 
+    def type_repr(self, t) -> str:
+        if isinstance(t, TimestampTZ):
+            return f"timestamp ({t.precision}) with time zone"
+        return super().type_repr(t)
+
 
 class PostgreSQL(ThreadedDatabase):
     dialect = PostgresqlDialect()

sqeleton/databases/snowflake.py

@@ -139,6 +139,11 @@ def set_timezone_to_utc(self) -> str:
     def optimizer_hints(self, hints: str) -> str:
         raise NotImplementedError("Optimizer hints not yet implemented in snowflake")
 
+    def type_repr(self, t) -> str:
+        if isinstance(t, TimestampTZ):
+            return f"timestamp_tz({t.precision})"
+        return super().type_repr(t)
+
 
 class Snowflake(Database):
     dialect = Dialect()

sqeleton/utils.py

@@ -322,3 +322,20 @@ def match_like(pattern: str, strs: Sequence[str]) -> Iterable[str]:
     for s in strs:
         if reo.match(s):
             yield s
+
+
+
+class UnknownMeta(type):
+    def __instancecheck__(self, instance):
+        return instance is Unknown
+
+    def __repr__(self):
+        return "Unknown"
+
+
+class Unknown(metaclass=UnknownMeta):
+    def __nonzero__(self):
+        raise TypeError()
+
+    def __new__(class_, *args, **kwargs):
+        raise RuntimeError("Unknown is a singleton")

tests/test_database.py

@@ -9,6 +9,7 @@
 from sqeleton.queries import table, current_timestamp, NormalizeAsString
 from .common import TEST_MYSQL_CONN_STRING
 from .common import str_to_checksum, test_each_database_in_list, get_conn, random_table_suffix
+from sqeleton.abcs.database_types import TimestampTZ
 
 TEST_DATABASES = {
     dbs.MySQL,
@@ -83,22 +84,22 @@ def test_current_timestamp(self):
     def test_correct_timezone(self):
         name = "tbl_" + random_table_suffix()
         db = get_conn(self.db_cls)
-        tbl = table(db.parse_table_name(name), schema={
-            "id": int, "created_at": "timestamp_tz(9)", "updated_at": "timestamp_tz(9)"
+        tbl = table(name, schema={
+            "id": int, "created_at": TimestampTZ(9), "updated_at": TimestampTZ(9)
         })
 
         db.query(tbl.create())
 
         tz = pytz.timezone('Europe/Berlin')
 
         now = datetime.now(tz)
-        db.query(table(db.parse_table_name(name)).insert_row("1", now, now))
+        db.query(table(name).insert_row("1", now, now))
         db.query(db.dialect.set_timezone_to_utc())
 
         t = db.table(name).query_schema()
-        t.schema["created_at"] = t.schema["created_at"].replace(precision=t.schema["created_at"].precision, rounds=True)
+        t.schema["created_at"] = t.schema["created_at"].replace(precision=t.schema["created_at"].precision)
 
-        tbl = table(db.parse_table_name(name), schema=t.schema)
+        tbl = table(name, schema=t.schema)
 
         results = db.query(tbl.select(NormalizeAsString(tbl[c]) for c in ["created_at", "updated_at"]), List[Tuple])