Apply typing.Literal whereever appropriate

py/server/deephaven/dbc/__init__.py

@@ -2,14 +2,14 @@
 #     Copyright (c) 2016-2022 Deephaven Data Labs and Patent Pending
 #
 """The dbc package includes the modules and functions for using external databases with Deephaven."""
-from typing import Any
+from typing import Any, Literal
 
 import deephaven.arrow as dharrow
 from deephaven import DHError
 from deephaven.table import Table
 
 
-def read_sql(conn: Any, query: str, driver: str = "connectorx") -> Table:
+def read_sql(conn: Any, query: str, driver: Literal["odbc", "adbc", "connectorx"] = "connectorx") -> Table:
     """Executes the provided SQL query via a supported driver and returns a Deephaven table.
 
     Args:

py/server/deephaven/table_listener.py

@@ -8,7 +8,7 @@
 from abc import ABC, abstractmethod
 from functools import wraps
 from inspect import signature
-from typing import Callable, Union, List, Generator, Dict, Optional
+from typing import Callable, Union, List, Generator, Dict, Optional, Literal
 
 import jpy
 import numpy
@@ -237,7 +237,8 @@ def modified_columns(self) -> List[str]:
         return list(cols) if cols else []
 
 
-def _do_locked(ug: Union[UpdateGraph, Table], f: Callable, lock_type="shared") -> None:
+def _do_locked(ug: Union[UpdateGraph, Table], f: Callable, lock_type: Literal["shared","exclusive"] = "shared") -> \
+        None:
     """Executes a function while holding the UpdateGraph (UG) lock.  Holding the UG lock
     ensures that the contents of a table will not change during a computation, but holding
     the lock also prevents table updates from happening.  The lock should be held for as little
@@ -308,7 +309,7 @@ def _wrap_listener_obj(t: Table, listener: TableListener):
 
 
 def listen(t: Table, listener: Union[Callable, TableListener], description: str = None, do_replay: bool = False,
-           replay_lock: str = "shared"):
+           replay_lock: Literal["shared", "exclusive"] = "shared"):
     """This is a convenience function that creates a TableListenerHandle object and immediately starts it to listen
     for table updates.
 
@@ -372,7 +373,7 @@ def __init__(self, t: Table, listener: Union[Callable, TableListener], descripti
 
         self.started = False
 
-    def start(self, do_replay: bool = False, replay_lock: str = "shared") -> None:
+    def start(self, do_replay: bool = False, replay_lock: Literal["shared", "exclusive"] = "shared") -> None:
         """Start the listener by registering it with the table and listening for updates.
 
         Args:

py/server/deephaven/time.py

@@ -7,7 +7,7 @@
 from __future__ import annotations
 
 import datetime
-from typing import Union, Optional
+from typing import Union, Optional, Literal
 
 import jpy
 import numpy
@@ -32,7 +32,7 @@
 # region Clock
 
 
-def dh_now(system: bool = False, resolution: str = 'ns') -> Instant:
+def dh_now(system: bool = False, resolution: Literal["ns", "ms"] = "ns") -> Instant:
     """ Provides the current datetime according to the current Deephaven clock.
 
     Query strings should use the built-in "now" function instead of this function.
@@ -43,9 +43,8 @@ def dh_now(system: bool = False, resolution: str = 'ns') -> Instant:
         system (bool): True to use the system clock; False to use the default clock.  Under most circumstances,
             the default clock will return the current system time, but during replay simulations, the default
             clock can return the replay time.
-
-        resolution (str): The resolution of the returned time.  The default 'ns' will return nanosecond resolution times
-            if possible. 'ms' will return millisecond resolution times.
+        resolution (str): The resolution of the returned time.  The default "ns" will return nanosecond resolution times
+            if possible. "ms" will return millisecond resolution times.
 
     Returns:
         Instant