Add generic options and remove schema options

src/snowflake/sqlalchemy/__init__.py

@@ -62,14 +62,24 @@
     VARIANT,
 )
 from .sql.custom_schema import DynamicTable, HybridTable
-from .sql.custom_schema.options import AsQuery, TargetLag, TimeUnit, Warehouse
+from .sql.custom_schema.options import (
+    AsQueryOption,
+    IdentifierOption,
+    KeywordOption,
+    LiteralOption,
+    SnowflakeKeyword,
+    TableOptionKey,
+    TargetLagOption,
+    TimeUnit,
+)
 from .util import _url as URL
 
 base.dialect = dialect = snowdialect.dialect
 
 __version__ = importlib_metadata.version("snowflake-sqlalchemy")
 
 __all__ = (
+    # Custom Types
     "BIGINT",
     "BINARY",
     "BOOLEAN",
@@ -104,6 +114,7 @@
     "TINYINT",
     "VARBINARY",
     "VARIANT",
+    # Custom Commands
     "MergeInto",
     "CSVFormatter",
     "JSONFormatter",
@@ -115,10 +126,17 @@
     "ExternalStage",
     "CreateStage",
     "CreateFileFormat",
+    # Custom Tables
+    "HybridTable",
     "DynamicTable",
-    "AsQuery",
-    "TargetLag",
+    # Custom Schema Options
+    "AsQueryOption",
+    "TargetLagOption",
+    "LiteralOption",
+    "IdentifierOption",
+    "KeywordOption",
+    # Enums
     "TimeUnit",
-    "Warehouse",
-    "HybridTable",
+    "TableOptionKey",
+    "SnowflakeKeyword",
 )

src/snowflake/sqlalchemy/base.py

@@ -5,6 +5,7 @@
 import itertools
 import operator
 import re
+from typing import List
 
 from sqlalchemy import exc as sa_exc
 from sqlalchemy import inspect, sql
@@ -26,8 +27,13 @@
     ExternalStage,
 )
 
+from .exc import (
+    CustomOptionsAreOnlySupportedOnSnowflakeTables,
+    UnexpectedOptionTypeError,
+)
 from .functions import flatten
-from .sql.custom_schema.options.table_option_base import TableOptionBase
+from .sql.custom_schema.custom_table_base import CustomTableBase
+from .sql.custom_schema.options.table_option import TableOption
 from .util import (
     _find_left_clause_to_join_from,
     _set_connection_interpolate_empty_sequences,
@@ -925,16 +931,24 @@ def handle_cluster_by(self, table):
 
     def post_create_table(self, table):
         text = self.handle_cluster_by(table)
-        options = [
-            option
-            for _, option in table.dialect_options[DIALECT_NAME].items()
-            if isinstance(option, TableOptionBase)
-        ]
-        options.sort(
-            key=lambda x: (x.__priority__.value, x.__option_name__), reverse=True
-        )
-        for option in options:
-            text += "\t" + option.render_option(self)
+        options = []
+        invalid_options: List[str] = []
+
+        for key, option in table.dialect_options[DIALECT_NAME].items():
+            if isinstance(option, TableOption):
+                options.append(option)
+            elif key not in ["clusterby", "*"]:
+                invalid_options.append(key)
+
+        if len(invalid_options) > 0:
+            raise UnexpectedOptionTypeError(sorted(invalid_options))
+
+        if isinstance(table, CustomTableBase):
+            options.sort(key=lambda x: (x.priority.value, x.option_name), reverse=True)
+            for option in options:
+                text += "\t" + option.render_option(self)
+        elif len(options) > 0:
+            raise CustomOptionsAreOnlySupportedOnSnowflakeTables()
 
         return text
 

src/snowflake/sqlalchemy/exc.py

@@ -0,0 +1,74 @@
+#
+# Copyright (c) 2012-2023 Snowflake Computing Inc. All rights reserved.
+
+from typing import List
+
+from sqlalchemy.exc import ArgumentError
+
+
+class NoPrimaryKeyError(ArgumentError):
+    def __init__(self, target: str):
+        super().__init__(f"Table {target} required primary key.")
+
+
+class UnsupportedPrimaryKeysAndForeignKeysError(ArgumentError):
+    def __init__(self, target: str):
+        super().__init__(f"Primary key and foreign keys are not supported in {target}.")
+
+
+class RequiredParametersNotProvidedError(ArgumentError):
+    def __init__(self, target: str, parameters: List[str]):
+        super().__init__(
+            f"{target} requires the following parameters: %s." % ", ".join(parameters)
+        )
+
+
+class UnexpectedTableOptionKeyError(ArgumentError):
+    def __init__(self, expected: str, actual: str):
+        super().__init__(f"Expected table option {expected} but got {actual}.")
+
+
+class OptionKeyNotProvidedError(ArgumentError):
+    def __init__(self, target: str):
+        super().__init__(
+            f"Expected option key in {target} option but got NoneType instead."
+        )
+
+
+class UnexpectedOptionParameterTypeError(ArgumentError):
+    def __init__(self, parameter_name: str, target: str, types: List[str]):
+        super().__init__(
+            f"Parameter {parameter_name} of {target} requires to be one"
+            f" of following types: {', '.join(types)}."
+        )
+
+
+class CustomOptionsAreOnlySupportedOnSnowflakeTables(ArgumentError):
+    def __init__(self):
+        super().__init__(
+            "Identifier, Literal, TargetLag and other custom options are only supported on Snowflake tables."
+        )
+
+
+class UnexpectedOptionTypeError(ArgumentError):
+    def __init__(self, options: List[str]):
+        super().__init__(
+            f"The following options are either unsupported or should be defined using a Snowflake table: {', '.join(options)}."
+        )
+
+
+class InvalidTableParameterTypeError(ArgumentError):
+    def __init__(self, name: str, input_type: str, expected_types: List[str]):
+        expected_types_str = "', '".join(expected_types)
+        super().__init__(
+            f"Invalid parameter type '{input_type}' provided for '{name}'. "
+            f"Expected one of the following types: '{expected_types_str}'.\n"
+        )
+
+
+class MultipleErrors(ArgumentError):
+    def __init__(self, errors):
+        self.errors = errors
+
+    def __str__(self):
+        return "\n ".join(str(e) for e in self.errors)

src/snowflake/sqlalchemy/sql/custom_schema/custom_table_base.py

@@ -2,21 +2,29 @@
 # Copyright (c) 2012-2023 Snowflake Computing Inc. All rights reserved.
 #
 import typing
-from typing import Any
+from typing import Any, List
 
-from sqlalchemy.exc import ArgumentError
 from sqlalchemy.sql.schema import MetaData, SchemaItem, Table
 
 from ..._constants import DIALECT_NAME
 from ...compat import IS_VERSION_20
 from ...custom_commands import NoneType
+from ...exc import (
+    MultipleErrors,
+    NoPrimaryKeyError,
+    RequiredParametersNotProvidedError,
+    UnsupportedPrimaryKeysAndForeignKeysError,
+)
 from .custom_table_prefix import CustomTablePrefix
-from .options.table_option import TableOption
+from .options.invalid_table_option import InvalidTableOption
+from .options.table_option import TableOption, TableOptionKey
 
 
 class CustomTableBase(Table):
     __table_prefixes__: typing.List[CustomTablePrefix] = []
     _support_primary_and_foreign_keys: bool = True
+    _enforce_primary_keys: bool = False
+    _required_parameters: List[TableOptionKey] = []
 
     @property
     def table_prefixes(self) -> typing.List[str]:
@@ -32,28 +40,74 @@ def __init__(
         if len(self.__table_prefixes__) > 0:
             prefixes = kw.get("prefixes", []) + self.table_prefixes
             kw.update(prefixes=prefixes)
+
         if not IS_VERSION_20 and hasattr(super(), "_init"):
+            kw.pop("_no_init", True)
             super()._init(name, metadata, *args, **kw)
         else:
             super().__init__(name, metadata, *args, **kw)
 
         if not kw.get("autoload_with", False):
             self._validate_table()
 
+    def _append_parameter_error(
+        self, parameter: str, expected_argument: str, current_argument: str
+    ) -> None:
+        if not hasattr(self, "_parameter_error"):
+            self._parameter_error = []
+        self._parameter_error.append((parameter, expected_argument, current_argument))
+
     def _validate_table(self):
+        exceptions: List[Exception] = []
+
+        for _, option in self.dialect_options[DIALECT_NAME].items():
+            if isinstance(option, InvalidTableOption):
+                exceptions.append(option.exception)
+
+        if isinstance(self.key, NoneType) and self._enforce_primary_keys:
+            exceptions.append(NoPrimaryKeyError(self.__class__.__name__))
+        missing_parameters: List[str] = []
+
+        for required_parameter in self._required_parameters:
+            if isinstance(self._get_dialect_option(required_parameter), NoneType):
+                missing_parameters.append(required_parameter.name.lower())
+        if missing_parameters:
+            exceptions.append(
+                RequiredParametersNotProvidedError(
+                    self.__class__.__name__, missing_parameters
+                )
+            )
+
         if not self._support_primary_and_foreign_keys and (
             self.primary_key or self.foreign_keys
         ):
-            raise ArgumentError(
-                f"Primary key and foreign keys are not supported in {' '.join(self.table_prefixes)} TABLE."
+            exceptions.append(
+                UnsupportedPrimaryKeysAndForeignKeysError(self.__class__.__name__)
             )
 
-        return True
+        if len(exceptions) > 1:
+            exceptions.sort(key=lambda e: str(e))
+            raise MultipleErrors(exceptions)
+        elif len(exceptions) == 1:
+            raise exceptions[0]
+
+    def _get_dialect_option(
+        self, option_name: TableOptionKey
+    ) -> typing.Optional[TableOption]:
+        if option_name.value in self.dialect_options[DIALECT_NAME]:
+            return self.dialect_options[DIALECT_NAME][option_name.value]
+        return None
 
-    def _get_dialect_option(self, option_name: str) -> typing.Optional[TableOption]:
-        if option_name in self.dialect_options[DIALECT_NAME]:
-            return self.dialect_options[DIALECT_NAME][option_name]
-        return NoneType
+    def _as_dialect_options(
+        self, table_options: List[TableOption]
+    ) -> typing.Dict[str, TableOption]:
+        result = {}
+        for table_option in table_options:
+            if isinstance(table_option, TableOption) and isinstance(
+                table_option.option_name, str
+            ):
+                result[DIALECT_NAME + "_" + table_option.option_name] = table_option
+        return result
 
     @classmethod
     def is_equal_type(cls, table: Table) -> bool: