Module astrapy.data.info.table_descriptor.table_altering
Expand source code
# Copyright DataStax, Inc.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
from __future__ import annotations
from abc import ABC, abstractmethod
from dataclasses import dataclass
from typing import Any, Dict, TypeVar, Union, cast
from astrapy.data.info.table_descriptor.table_columns import (
TableColumnTypeDescriptor,
)
from astrapy.data.info.vectorize import VectorServiceOptions
from astrapy.utils.parsing import _warn_residual_keys
ATO = TypeVar("ATO", bound="AlterTableOperation")
@dataclass
class AlterTableOperation(ABC):
"""
An abstract class representing a generic "alter table" operation. Concrete
implementations are used to represent operations such as adding/dropping columns
and adding/dropping the vectorize service (server-side embedding computations).
`AlterTableOperation` objects are the parameter to the Table's `alter` method.
Please consult the documentation of the concrete subclasses for more info.
"""
_name: str
@abstractmethod
def as_dict(self) -> dict[str, Any]: ...
@staticmethod
def from_full_dict(operation_dict: dict[str, Any]) -> AlterTableOperation:
"""
Inspect a provided dictionary and make it into the correct concrete subclass
of AlterTableOperation depending on its contents.
Note: while the nature of the operation must be the top-level single key of
the (nested) dictionary parameter to this method (such as "add" or
"dropVectorize"), the resulting `AlterTableOperation` object encodes the content
of the corresponding value. Likewise, calling the `as_dict()` method of
the result from this method does not return the whole original input, rather
the "one level in" part (see the example provided here).
Args:
operation_dict: a dictionary such as `{"add": ...}`, whose outermost *value*
corresponds to the desired operation.
Returns:
an `AlterTableOperation` object chosen after inspection of
the provided input.
Example:
>>> full_dict = {"drop": {"columns": ["col1", "col2"]}}
>>> alter_op = AlterTableOperation.from_full_dict(full_dict)
>>> alter_op
AlterTableDropColumns(columns=[col1,col2])
>>> alter_op.as_dict()
{'columns': ['col1', 'col2']}
>>> alter_op.as_dict() == full_dict["drop"]
True
"""
key_set = set(operation_dict.keys())
if key_set == {"add"}:
return AlterTableAddColumns.coerce(operation_dict["add"])
elif key_set == {"drop"}:
return AlterTableDropColumns.coerce(operation_dict["drop"])
elif key_set == {"addVectorize"}:
return AlterTableAddVectorize.coerce(operation_dict["addVectorize"])
elif key_set == {"dropVectorize"}:
return AlterTableDropVectorize.coerce(operation_dict["dropVectorize"])
else:
raise ValueError(
f"Cannot parse a dict with keys {', '.join(sorted(key_set))} "
"into an AlterTableOperation"
)
@classmethod
@abstractmethod
def coerce(cls: type[ATO], raw_input: ATO | dict[str, Any]) -> ATO: ...
@dataclass
class AlterTableAddColumns(AlterTableOperation):
"""
An object representing the alter-table operation of adding column(s),
for use as argument to the table's `alter()` method.
Attributes:
columns: a mapping between the names of the columns to add and
`TableColumnTypeDescriptor` objects, formatted in the same way as
the `columns` attribute of `CreateTableDefinition`.
"""
columns: dict[str, TableColumnTypeDescriptor]
def __init__(
self, columns: dict[str, TableColumnTypeDescriptor | dict[str, Any]]
) -> None:
self._name = "add"
self.columns = {
col_n: TableColumnTypeDescriptor.coerce(col_v)
for col_n, col_v in columns.items()
}
def __repr__(self) -> str:
_col_desc = f"columns=[{','.join(sorted(self.columns.keys()))}]"
return f"{self.__class__.__name__}({_col_desc})"
def as_dict(self) -> dict[str, Any]:
"""Recast this object into a dictionary."""
return {
"columns": {col_n: col_v.as_dict() for col_n, col_v in self.columns.items()}
}
@classmethod
def _from_dict(cls, raw_dict: dict[str, Any]) -> AlterTableAddColumns:
"""
Create an instance of AlterTableAddColumns from a dictionary
such as one suitable as (partial) command payload.
"""
_warn_residual_keys(cls, raw_dict, {"columns"})
return AlterTableAddColumns(
columns={
col_n: TableColumnTypeDescriptor.coerce(col_v)
for col_n, col_v in raw_dict["columns"].items()
},
)
@classmethod
def coerce(
cls, raw_input: AlterTableAddColumns | dict[str, Any]
) -> AlterTableAddColumns:
"""
Normalize the input, whether an object already or a plain dictionary
of the right structure, into an AlterTableAddColumns.
"""
if isinstance(raw_input, AlterTableAddColumns):
return raw_input
else:
return cls._from_dict(raw_input)
@dataclass
class AlterTableDropColumns(AlterTableOperation):
"""
An object representing the alter-table operation of dropping column(s),
for use as argument to the table's `alter()` method.
Attributes:
columns: a list of the column names to drop.
Passing a single string has the same effect as passing a single-item list.
"""
columns: list[str]
def __init__(self, columns: list[str] | str) -> None:
self._name = "drop"
self.columns = [columns] if isinstance(columns, str) else columns
def __repr__(self) -> str:
_col_desc = f"columns=[{','.join(self.columns)}]"
return f"{self.__class__.__name__}({_col_desc})"
def as_dict(self) -> dict[str, Any]:
"""Recast this object into a dictionary."""
return {
"columns": self.columns,
}
@classmethod
def _from_dict(cls, raw_dict: dict[str, Any]) -> AlterTableDropColumns:
"""
Create an instance of AlterTableDropColumns from a dictionary
such as one suitable as (partial) command payload.
"""
_warn_residual_keys(cls, raw_dict, {"columns"})
return AlterTableDropColumns(
columns=raw_dict["columns"],
)
@classmethod
def coerce(
cls, raw_input: AlterTableDropColumns | dict[str, Any]
) -> AlterTableDropColumns:
"""
Normalize the input, whether an object already or a plain dictionary
of the right structure, into an AlterTableDropColumns.
"""
if isinstance(raw_input, AlterTableDropColumns):
return raw_input
else:
return cls._from_dict(raw_input)
@dataclass
class AlterTableAddVectorize(AlterTableOperation):
"""
An object representing the alter-table operation of enabling the vectorize service
(i.e. server-side embedding computation) on one or more columns,
for use as argument to the table's `alter()` method.
Attributes:
columns: a mapping between column names and the corresponding
`VectorServiceOptions` objects describing the settings for the
desired vectorize service.
"""
columns: dict[str, VectorServiceOptions]
def __init__(
self, columns: dict[str, VectorServiceOptions | dict[str, Any]]
) -> None:
self._name = "addVectorize"
columns_ = {
col_n: VectorServiceOptions.coerce(col_v)
for col_n, col_v in columns.items()
}
if any(_col_svc is None for _col_svc in columns_.values()):
raise ValueError(
"Vector service definition cannot be None for AlterTableAddVectorize"
)
self.columns = cast(Dict[str, VectorServiceOptions], columns_)
def __repr__(self) -> str:
_cols_desc = [
f"{col_n}({col_svc.provider}/{col_svc.model_name})"
for col_n, col_svc in sorted(self.columns.items())
]
return f"{self.__class__.__name__}(columns={', '.join(_cols_desc)})"
def as_dict(self) -> dict[str, Any]:
"""Recast this object into a dictionary."""
return {
"columns": {
col_n: col_svc.as_dict() for col_n, col_svc in self.columns.items()
}
}
@classmethod
def _from_dict(cls, raw_dict: dict[str, Any]) -> AlterTableAddVectorize:
"""
Create an instance of AlterTableAddVectorize from a dictionary
such as one suitable as (partial) command payload.
"""
_warn_residual_keys(cls, raw_dict, {"columns"})
_columns: dict[str, VectorServiceOptions | None] = {
col_n: VectorServiceOptions.coerce(col_v)
for col_n, col_v in raw_dict["columns"].items()
}
if any(_col_svc is None for _col_svc in _columns.values()):
raise ValueError(
"Vector service definition cannot be None for AlterTableAddVectorize"
)
return AlterTableAddVectorize(
columns=cast(
Dict[str, Union[VectorServiceOptions, Dict[str, Any]]],
_columns,
)
)
@classmethod
def coerce(
cls, raw_input: AlterTableAddVectorize | dict[str, Any]
) -> AlterTableAddVectorize:
"""
Normalize the input, whether an object already or a plain dictionary
of the right structure, into an AlterTableAddVectorize.
"""
if isinstance(raw_input, AlterTableAddVectorize):
return raw_input
else:
return cls._from_dict(raw_input)
@dataclass
class AlterTableDropVectorize(AlterTableOperation):
"""
An object representing the alter-table operation of removing the vectorize
service (i.e. the server-side embedding computation) from one or more columns,
for use as argument to the table's `alter()` method.
Note: this operation does not drop the column, simply unsets its vectorize
service. Existing embedding vectors, stored in the table, are retained.
Attributes:
columns: a list of the column names whose vectorize service is to be removed.
Passing a single string has the same effect as passing a single-item list.
"""
columns: list[str]
def __init__(self, columns: list[str] | str) -> None:
self._name = "dropVectorize"
self.columns = [columns] if isinstance(columns, str) else columns
def __repr__(self) -> str:
_col_desc = f"columns=[{','.join(self.columns)}]"
return f"{self.__class__.__name__}({_col_desc})"
def as_dict(self) -> dict[str, Any]:
"""Recast this object into a dictionary."""
return {
"columns": self.columns,
}
@classmethod
def _from_dict(cls, raw_dict: dict[str, Any]) -> AlterTableDropVectorize:
"""
Create an instance of AlterTableDropVectorize from a dictionary
such as one suitable as (partial) command payload.
"""
_warn_residual_keys(cls, raw_dict, {"columns"})
return AlterTableDropVectorize(
columns=raw_dict["columns"],
)
@classmethod
def coerce(
cls, raw_input: AlterTableDropVectorize | dict[str, Any]
) -> AlterTableDropVectorize:
"""
Normalize the input, whether an object already or a plain dictionary
of the right structure, into an AlterTableDropVectorize.
"""
if isinstance(raw_input, AlterTableDropVectorize):
return raw_input
else:
return cls._from_dict(raw_input)
Classes
class AlterTableAddColumns (columns: dict[str, TableColumnTypeDescriptor | dict[str, Any]])
-
An object representing the alter-table operation of adding column(s), for use as argument to the table's
alter()
method.Attributes
columns
- a mapping between the names of the columns to add and
TableColumnTypeDescriptor
objects, formatted in the same way as thecolumns
attribute ofCreateTableDefinition
.
Expand source code
@dataclass class AlterTableAddColumns(AlterTableOperation): """ An object representing the alter-table operation of adding column(s), for use as argument to the table's `alter()` method. Attributes: columns: a mapping between the names of the columns to add and `TableColumnTypeDescriptor` objects, formatted in the same way as the `columns` attribute of `CreateTableDefinition`. """ columns: dict[str, TableColumnTypeDescriptor] def __init__( self, columns: dict[str, TableColumnTypeDescriptor | dict[str, Any]] ) -> None: self._name = "add" self.columns = { col_n: TableColumnTypeDescriptor.coerce(col_v) for col_n, col_v in columns.items() } def __repr__(self) -> str: _col_desc = f"columns=[{','.join(sorted(self.columns.keys()))}]" return f"{self.__class__.__name__}({_col_desc})" def as_dict(self) -> dict[str, Any]: """Recast this object into a dictionary.""" return { "columns": {col_n: col_v.as_dict() for col_n, col_v in self.columns.items()} } @classmethod def _from_dict(cls, raw_dict: dict[str, Any]) -> AlterTableAddColumns: """ Create an instance of AlterTableAddColumns from a dictionary such as one suitable as (partial) command payload. """ _warn_residual_keys(cls, raw_dict, {"columns"}) return AlterTableAddColumns( columns={ col_n: TableColumnTypeDescriptor.coerce(col_v) for col_n, col_v in raw_dict["columns"].items() }, ) @classmethod def coerce( cls, raw_input: AlterTableAddColumns | dict[str, Any] ) -> AlterTableAddColumns: """ Normalize the input, whether an object already or a plain dictionary of the right structure, into an AlterTableAddColumns. """ if isinstance(raw_input, AlterTableAddColumns): return raw_input else: return cls._from_dict(raw_input)
Ancestors
- AlterTableOperation
- abc.ABC
Class variables
var columns : dict[str, TableColumnTypeDescriptor]
Static methods
def coerce(raw_input: AlterTableAddColumns | dict[str, Any]) ‑> AlterTableAddColumns
-
Normalize the input, whether an object already or a plain dictionary of the right structure, into an AlterTableAddColumns.
Expand source code
@classmethod def coerce( cls, raw_input: AlterTableAddColumns | dict[str, Any] ) -> AlterTableAddColumns: """ Normalize the input, whether an object already or a plain dictionary of the right structure, into an AlterTableAddColumns. """ if isinstance(raw_input, AlterTableAddColumns): return raw_input else: return cls._from_dict(raw_input)
Methods
def as_dict(self) ‑> dict[str, typing.Any]
-
Recast this object into a dictionary.
Expand source code
def as_dict(self) -> dict[str, Any]: """Recast this object into a dictionary.""" return { "columns": {col_n: col_v.as_dict() for col_n, col_v in self.columns.items()} }
Inherited members
class AlterTableAddVectorize (columns: dict[str, VectorServiceOptions | dict[str, Any]])
-
An object representing the alter-table operation of enabling the vectorize service (i.e. server-side embedding computation) on one or more columns, for use as argument to the table's
alter()
method.Attributes
columns
- a mapping between column names and the corresponding
VectorServiceOptions
objects describing the settings for the desired vectorize service.
Expand source code
@dataclass class AlterTableAddVectorize(AlterTableOperation): """ An object representing the alter-table operation of enabling the vectorize service (i.e. server-side embedding computation) on one or more columns, for use as argument to the table's `alter()` method. Attributes: columns: a mapping between column names and the corresponding `VectorServiceOptions` objects describing the settings for the desired vectorize service. """ columns: dict[str, VectorServiceOptions] def __init__( self, columns: dict[str, VectorServiceOptions | dict[str, Any]] ) -> None: self._name = "addVectorize" columns_ = { col_n: VectorServiceOptions.coerce(col_v) for col_n, col_v in columns.items() } if any(_col_svc is None for _col_svc in columns_.values()): raise ValueError( "Vector service definition cannot be None for AlterTableAddVectorize" ) self.columns = cast(Dict[str, VectorServiceOptions], columns_) def __repr__(self) -> str: _cols_desc = [ f"{col_n}({col_svc.provider}/{col_svc.model_name})" for col_n, col_svc in sorted(self.columns.items()) ] return f"{self.__class__.__name__}(columns={', '.join(_cols_desc)})" def as_dict(self) -> dict[str, Any]: """Recast this object into a dictionary.""" return { "columns": { col_n: col_svc.as_dict() for col_n, col_svc in self.columns.items() } } @classmethod def _from_dict(cls, raw_dict: dict[str, Any]) -> AlterTableAddVectorize: """ Create an instance of AlterTableAddVectorize from a dictionary such as one suitable as (partial) command payload. """ _warn_residual_keys(cls, raw_dict, {"columns"}) _columns: dict[str, VectorServiceOptions | None] = { col_n: VectorServiceOptions.coerce(col_v) for col_n, col_v in raw_dict["columns"].items() } if any(_col_svc is None for _col_svc in _columns.values()): raise ValueError( "Vector service definition cannot be None for AlterTableAddVectorize" ) return AlterTableAddVectorize( columns=cast( Dict[str, Union[VectorServiceOptions, Dict[str, Any]]], _columns, ) ) @classmethod def coerce( cls, raw_input: AlterTableAddVectorize | dict[str, Any] ) -> AlterTableAddVectorize: """ Normalize the input, whether an object already or a plain dictionary of the right structure, into an AlterTableAddVectorize. """ if isinstance(raw_input, AlterTableAddVectorize): return raw_input else: return cls._from_dict(raw_input)
Ancestors
- AlterTableOperation
- abc.ABC
Class variables
var columns : dict[str, VectorServiceOptions]
Static methods
def coerce(raw_input: AlterTableAddVectorize | dict[str, Any]) ‑> AlterTableAddVectorize
-
Normalize the input, whether an object already or a plain dictionary of the right structure, into an AlterTableAddVectorize.
Expand source code
@classmethod def coerce( cls, raw_input: AlterTableAddVectorize | dict[str, Any] ) -> AlterTableAddVectorize: """ Normalize the input, whether an object already or a plain dictionary of the right structure, into an AlterTableAddVectorize. """ if isinstance(raw_input, AlterTableAddVectorize): return raw_input else: return cls._from_dict(raw_input)
Methods
def as_dict(self) ‑> dict[str, typing.Any]
-
Recast this object into a dictionary.
Expand source code
def as_dict(self) -> dict[str, Any]: """Recast this object into a dictionary.""" return { "columns": { col_n: col_svc.as_dict() for col_n, col_svc in self.columns.items() } }
Inherited members
class AlterTableDropColumns (columns: list[str] | str)
-
An object representing the alter-table operation of dropping column(s), for use as argument to the table's
alter()
method.Attributes
columns
- a list of the column names to drop. Passing a single string has the same effect as passing a single-item list.
Expand source code
@dataclass class AlterTableDropColumns(AlterTableOperation): """ An object representing the alter-table operation of dropping column(s), for use as argument to the table's `alter()` method. Attributes: columns: a list of the column names to drop. Passing a single string has the same effect as passing a single-item list. """ columns: list[str] def __init__(self, columns: list[str] | str) -> None: self._name = "drop" self.columns = [columns] if isinstance(columns, str) else columns def __repr__(self) -> str: _col_desc = f"columns=[{','.join(self.columns)}]" return f"{self.__class__.__name__}({_col_desc})" def as_dict(self) -> dict[str, Any]: """Recast this object into a dictionary.""" return { "columns": self.columns, } @classmethod def _from_dict(cls, raw_dict: dict[str, Any]) -> AlterTableDropColumns: """ Create an instance of AlterTableDropColumns from a dictionary such as one suitable as (partial) command payload. """ _warn_residual_keys(cls, raw_dict, {"columns"}) return AlterTableDropColumns( columns=raw_dict["columns"], ) @classmethod def coerce( cls, raw_input: AlterTableDropColumns | dict[str, Any] ) -> AlterTableDropColumns: """ Normalize the input, whether an object already or a plain dictionary of the right structure, into an AlterTableDropColumns. """ if isinstance(raw_input, AlterTableDropColumns): return raw_input else: return cls._from_dict(raw_input)
Ancestors
- AlterTableOperation
- abc.ABC
Class variables
var columns : list[str]
Static methods
def coerce(raw_input: AlterTableDropColumns | dict[str, Any]) ‑> AlterTableDropColumns
-
Normalize the input, whether an object already or a plain dictionary of the right structure, into an AlterTableDropColumns.
Expand source code
@classmethod def coerce( cls, raw_input: AlterTableDropColumns | dict[str, Any] ) -> AlterTableDropColumns: """ Normalize the input, whether an object already or a plain dictionary of the right structure, into an AlterTableDropColumns. """ if isinstance(raw_input, AlterTableDropColumns): return raw_input else: return cls._from_dict(raw_input)
Methods
def as_dict(self) ‑> dict[str, typing.Any]
-
Recast this object into a dictionary.
Expand source code
def as_dict(self) -> dict[str, Any]: """Recast this object into a dictionary.""" return { "columns": self.columns, }
Inherited members
class AlterTableDropVectorize (columns: list[str] | str)
-
An object representing the alter-table operation of removing the vectorize service (i.e. the server-side embedding computation) from one or more columns, for use as argument to the table's
alter()
method.Note: this operation does not drop the column, simply unsets its vectorize service. Existing embedding vectors, stored in the table, are retained.
Attributes
columns
- a list of the column names whose vectorize service is to be removed. Passing a single string has the same effect as passing a single-item list.
Expand source code
@dataclass class AlterTableDropVectorize(AlterTableOperation): """ An object representing the alter-table operation of removing the vectorize service (i.e. the server-side embedding computation) from one or more columns, for use as argument to the table's `alter()` method. Note: this operation does not drop the column, simply unsets its vectorize service. Existing embedding vectors, stored in the table, are retained. Attributes: columns: a list of the column names whose vectorize service is to be removed. Passing a single string has the same effect as passing a single-item list. """ columns: list[str] def __init__(self, columns: list[str] | str) -> None: self._name = "dropVectorize" self.columns = [columns] if isinstance(columns, str) else columns def __repr__(self) -> str: _col_desc = f"columns=[{','.join(self.columns)}]" return f"{self.__class__.__name__}({_col_desc})" def as_dict(self) -> dict[str, Any]: """Recast this object into a dictionary.""" return { "columns": self.columns, } @classmethod def _from_dict(cls, raw_dict: dict[str, Any]) -> AlterTableDropVectorize: """ Create an instance of AlterTableDropVectorize from a dictionary such as one suitable as (partial) command payload. """ _warn_residual_keys(cls, raw_dict, {"columns"}) return AlterTableDropVectorize( columns=raw_dict["columns"], ) @classmethod def coerce( cls, raw_input: AlterTableDropVectorize | dict[str, Any] ) -> AlterTableDropVectorize: """ Normalize the input, whether an object already or a plain dictionary of the right structure, into an AlterTableDropVectorize. """ if isinstance(raw_input, AlterTableDropVectorize): return raw_input else: return cls._from_dict(raw_input)
Ancestors
- AlterTableOperation
- abc.ABC
Class variables
var columns : list[str]
Static methods
def coerce(raw_input: AlterTableDropVectorize | dict[str, Any]) ‑> AlterTableDropVectorize
-
Normalize the input, whether an object already or a plain dictionary of the right structure, into an AlterTableDropVectorize.
Expand source code
@classmethod def coerce( cls, raw_input: AlterTableDropVectorize | dict[str, Any] ) -> AlterTableDropVectorize: """ Normalize the input, whether an object already or a plain dictionary of the right structure, into an AlterTableDropVectorize. """ if isinstance(raw_input, AlterTableDropVectorize): return raw_input else: return cls._from_dict(raw_input)
Methods
def as_dict(self) ‑> dict[str, typing.Any]
-
Recast this object into a dictionary.
Expand source code
def as_dict(self) -> dict[str, Any]: """Recast this object into a dictionary.""" return { "columns": self.columns, }
Inherited members
class AlterTableOperation (_name: str)
-
An abstract class representing a generic "alter table" operation. Concrete implementations are used to represent operations such as adding/dropping columns and adding/dropping the vectorize service (server-side embedding computations).
AlterTableOperation
objects are the parameter to the Table'salter
method.Please consult the documentation of the concrete subclasses for more info.
Expand source code
@dataclass class AlterTableOperation(ABC): """ An abstract class representing a generic "alter table" operation. Concrete implementations are used to represent operations such as adding/dropping columns and adding/dropping the vectorize service (server-side embedding computations). `AlterTableOperation` objects are the parameter to the Table's `alter` method. Please consult the documentation of the concrete subclasses for more info. """ _name: str @abstractmethod def as_dict(self) -> dict[str, Any]: ... @staticmethod def from_full_dict(operation_dict: dict[str, Any]) -> AlterTableOperation: """ Inspect a provided dictionary and make it into the correct concrete subclass of AlterTableOperation depending on its contents. Note: while the nature of the operation must be the top-level single key of the (nested) dictionary parameter to this method (such as "add" or "dropVectorize"), the resulting `AlterTableOperation` object encodes the content of the corresponding value. Likewise, calling the `as_dict()` method of the result from this method does not return the whole original input, rather the "one level in" part (see the example provided here). Args: operation_dict: a dictionary such as `{"add": ...}`, whose outermost *value* corresponds to the desired operation. Returns: an `AlterTableOperation` object chosen after inspection of the provided input. Example: >>> full_dict = {"drop": {"columns": ["col1", "col2"]}} >>> alter_op = AlterTableOperation.from_full_dict(full_dict) >>> alter_op AlterTableDropColumns(columns=[col1,col2]) >>> alter_op.as_dict() {'columns': ['col1', 'col2']} >>> alter_op.as_dict() == full_dict["drop"] True """ key_set = set(operation_dict.keys()) if key_set == {"add"}: return AlterTableAddColumns.coerce(operation_dict["add"]) elif key_set == {"drop"}: return AlterTableDropColumns.coerce(operation_dict["drop"]) elif key_set == {"addVectorize"}: return AlterTableAddVectorize.coerce(operation_dict["addVectorize"]) elif key_set == {"dropVectorize"}: return AlterTableDropVectorize.coerce(operation_dict["dropVectorize"]) else: raise ValueError( f"Cannot parse a dict with keys {', '.join(sorted(key_set))} " "into an AlterTableOperation" ) @classmethod @abstractmethod def coerce(cls: type[ATO], raw_input: ATO | dict[str, Any]) -> ATO: ...
Ancestors
- abc.ABC
Subclasses
Static methods
def coerce(raw_input: ATO | dict[str, Any]) ‑> ~ATO
-
Expand source code
@classmethod @abstractmethod def coerce(cls: type[ATO], raw_input: ATO | dict[str, Any]) -> ATO: ...
def from_full_dict(operation_dict: dict[str, Any]) ‑> AlterTableOperation
-
Inspect a provided dictionary and make it into the correct concrete subclass of AlterTableOperation depending on its contents.
Note: while the nature of the operation must be the top-level single key of the (nested) dictionary parameter to this method (such as "add" or "dropVectorize"), the resulting
AlterTableOperation
object encodes the content of the corresponding value. Likewise, calling theas_dict()
method of the result from this method does not return the whole original input, rather the "one level in" part (see the example provided here).Args
operation_dict
- a dictionary such as
{"add": ...}
, whose outermost value
corresponds to the desired operation.
Returns
an
AlterTableOperation
object chosen after inspection of the provided input.Example
>>> full_dict = {"drop": {"columns": ["col1", "col2"]}} >>> alter_op = AlterTableOperation.from_full_dict(full_dict) >>> alter_op AlterTableDropColumns(columns=[col1,col2]) >>> alter_op.as_dict() {'columns': ['col1', 'col2']} >>> alter_op.as_dict() == full_dict["drop"] True
Expand source code
@staticmethod def from_full_dict(operation_dict: dict[str, Any]) -> AlterTableOperation: """ Inspect a provided dictionary and make it into the correct concrete subclass of AlterTableOperation depending on its contents. Note: while the nature of the operation must be the top-level single key of the (nested) dictionary parameter to this method (such as "add" or "dropVectorize"), the resulting `AlterTableOperation` object encodes the content of the corresponding value. Likewise, calling the `as_dict()` method of the result from this method does not return the whole original input, rather the "one level in" part (see the example provided here). Args: operation_dict: a dictionary such as `{"add": ...}`, whose outermost *value* corresponds to the desired operation. Returns: an `AlterTableOperation` object chosen after inspection of the provided input. Example: >>> full_dict = {"drop": {"columns": ["col1", "col2"]}} >>> alter_op = AlterTableOperation.from_full_dict(full_dict) >>> alter_op AlterTableDropColumns(columns=[col1,col2]) >>> alter_op.as_dict() {'columns': ['col1', 'col2']} >>> alter_op.as_dict() == full_dict["drop"] True """ key_set = set(operation_dict.keys()) if key_set == {"add"}: return AlterTableAddColumns.coerce(operation_dict["add"]) elif key_set == {"drop"}: return AlterTableDropColumns.coerce(operation_dict["drop"]) elif key_set == {"addVectorize"}: return AlterTableAddVectorize.coerce(operation_dict["addVectorize"]) elif key_set == {"dropVectorize"}: return AlterTableDropVectorize.coerce(operation_dict["dropVectorize"]) else: raise ValueError( f"Cannot parse a dict with keys {', '.join(sorted(key_set))} " "into an AlterTableOperation" )
Methods
def as_dict(self) ‑> dict[str, typing.Any]
-
Expand source code
@abstractmethod def as_dict(self) -> dict[str, Any]: ...