sqlglot.schema

  1from __future__ import annotations
  2
  3import abc
  4import typing as t
  5
  6from sqlglot import expressions as exp
  7from sqlglot.dialects.dialect import Dialect
  8from sqlglot.errors import SchemaError
  9from sqlglot.helper import dict_depth, first
 10from sqlglot.trie import TrieResult, in_trie, new_trie
 11
 12if t.TYPE_CHECKING:
 13    from sqlglot.dialects.dialect import DialectType
 14
 15    ColumnMapping = t.Union[t.Dict, str, t.List]
 16
 17
 18class Schema(abc.ABC):
 19    """Abstract base class for database schemas"""
 20
 21    @property
 22    def dialect(self) -> t.Optional[Dialect]:
 23        """
 24        Returns None by default. Subclasses that require dialect-specific
 25        behavior should override this property.
 26        """
 27        return None
 28
 29    @abc.abstractmethod
 30    def add_table(
 31        self,
 32        table: exp.Table | str,
 33        column_mapping: t.Optional[ColumnMapping] = None,
 34        dialect: DialectType = None,
 35        normalize: t.Optional[bool] = None,
 36        match_depth: bool = True,
 37    ) -> None:
 38        """
 39        Register or update a table. Some implementing classes may require column information to also be provided.
 40        The added table must have the necessary number of qualifiers in its path to match the schema's nesting level.
 41
 42        Args:
 43            table: the `Table` expression instance or string representing the table.
 44            column_mapping: a column mapping that describes the structure of the table.
 45            dialect: the SQL dialect that will be used to parse `table` if it's a string.
 46            normalize: whether to normalize identifiers according to the dialect of interest.
 47            match_depth: whether to enforce that the table must match the schema's depth or not.
 48        """
 49
 50    @abc.abstractmethod
 51    def column_names(
 52        self,
 53        table: exp.Table | str,
 54        only_visible: bool = False,
 55        dialect: DialectType = None,
 56        normalize: t.Optional[bool] = None,
 57    ) -> t.Sequence[str]:
 58        """
 59        Get the column names for a table.
 60
 61        Args:
 62            table: the `Table` expression instance.
 63            only_visible: whether to include invisible columns.
 64            dialect: the SQL dialect that will be used to parse `table` if it's a string.
 65            normalize: whether to normalize identifiers according to the dialect of interest.
 66
 67        Returns:
 68            The sequence of column names.
 69        """
 70
 71    @abc.abstractmethod
 72    def get_column_type(
 73        self,
 74        table: exp.Table | str,
 75        column: exp.Column | str,
 76        dialect: DialectType = None,
 77        normalize: t.Optional[bool] = None,
 78    ) -> exp.DataType:
 79        """
 80        Get the `sqlglot.exp.DataType` type of a column in the schema.
 81
 82        Args:
 83            table: the source table.
 84            column: the target column.
 85            dialect: the SQL dialect that will be used to parse `table` if it's a string.
 86            normalize: whether to normalize identifiers according to the dialect of interest.
 87
 88        Returns:
 89            The resulting column type.
 90        """
 91
 92    def has_column(
 93        self,
 94        table: exp.Table | str,
 95        column: exp.Column | str,
 96        dialect: DialectType = None,
 97        normalize: t.Optional[bool] = None,
 98    ) -> bool:
 99        """
100        Returns whether `column` appears in `table`'s schema.
101
102        Args:
103            table: the source table.
104            column: the target column.
105            dialect: the SQL dialect that will be used to parse `table` if it's a string.
106            normalize: whether to normalize identifiers according to the dialect of interest.
107
108        Returns:
109            True if the column appears in the schema, False otherwise.
110        """
111        name = column if isinstance(column, str) else column.name
112        return name in self.column_names(table, dialect=dialect, normalize=normalize)
113
114    def get_udf_type(
115        self,
116        udf: exp.Anonymous | str,
117        dialect: DialectType = None,
118        normalize: t.Optional[bool] = None,
119    ) -> exp.DataType:
120        """
121        Get the return type of a UDF.
122
123        Args:
124            udf: the UDF expression or string.
125            dialect: the SQL dialect for parsing string arguments.
126            normalize: whether to normalize identifiers.
127
128        Returns:
129            The return type as a DataType, or UNKNOWN if not found.
130        """
131        return exp.DataType.build("unknown")
132
133    @property
134    @abc.abstractmethod
135    def supported_table_args(self) -> t.Tuple[str, ...]:
136        """
137        Table arguments this schema support, e.g. `("this", "db", "catalog")`
138        """
139
140    @property
141    def empty(self) -> bool:
142        """Returns whether the schema is empty."""
143        return True
144
145
146class AbstractMappingSchema:
147    def __init__(
148        self,
149        mapping: t.Optional[t.Dict] = None,
150        udf_mapping: t.Optional[t.Dict] = None,
151    ) -> None:
152        self.mapping = mapping or {}
153        self.mapping_trie = new_trie(
154            tuple(reversed(t)) for t in flatten_schema(self.mapping, depth=self.depth())
155        )
156
157        self.udf_mapping = udf_mapping or {}
158        self.udf_trie = new_trie(
159            tuple(reversed(t)) for t in flatten_schema(self.udf_mapping, depth=self.udf_depth())
160        )
161
162        self._supported_table_args: t.Tuple[str, ...] = tuple()
163
164    @property
165    def empty(self) -> bool:
166        return not self.mapping
167
168    def depth(self) -> int:
169        return dict_depth(self.mapping)
170
171    def udf_depth(self) -> int:
172        return dict_depth(self.udf_mapping)
173
174    @property
175    def supported_table_args(self) -> t.Tuple[str, ...]:
176        if not self._supported_table_args and self.mapping:
177            depth = self.depth()
178
179            if not depth:  # None
180                self._supported_table_args = tuple()
181            elif 1 <= depth <= 3:
182                self._supported_table_args = exp.TABLE_PARTS[:depth]
183            else:
184                raise SchemaError(f"Invalid mapping shape. Depth: {depth}")
185
186        return self._supported_table_args
187
188    def table_parts(self, table: exp.Table) -> t.List[str]:
189        return [p.name for p in reversed(table.parts)]
190
191    def udf_parts(self, udf: exp.Anonymous) -> t.List[str]:
192        # a.b.c(...) is represented as Dot(Dot(a, b), Anonymous(c, ...))
193        parent = udf.parent
194        parts = [p.name for p in parent.flatten()] if isinstance(parent, exp.Dot) else [udf.name]
195        return list(reversed(parts))[0 : self.udf_depth()]
196
197    def _find_in_trie(
198        self,
199        parts: t.List[str],
200        trie: t.Dict,
201        raise_on_missing: bool,
202    ) -> t.Optional[t.List[str]]:
203        value, trie = in_trie(trie, parts)
204
205        if value == TrieResult.FAILED:
206            return None
207
208        if value == TrieResult.PREFIX:
209            possibilities = flatten_schema(trie)
210
211            if len(possibilities) == 1:
212                parts.extend(possibilities[0])
213            else:
214                if raise_on_missing:
215                    joined_parts = ".".join(parts)
216                    message = ", ".join(".".join(p) for p in possibilities)
217                    raise SchemaError(f"Ambiguous mapping for {joined_parts}: {message}.")
218
219                return None
220
221        return parts
222
223    def find(
224        self, table: exp.Table, raise_on_missing: bool = True, ensure_data_types: bool = False
225    ) -> t.Optional[t.Any]:
226        """
227        Returns the schema of a given table.
228
229        Args:
230            table: the target table.
231            raise_on_missing: whether to raise in case the schema is not found.
232            ensure_data_types: whether to convert `str` types to their `DataType` equivalents.
233
234        Returns:
235            The schema of the target table.
236        """
237        parts = self.table_parts(table)[0 : len(self.supported_table_args)]
238        resolved_parts = self._find_in_trie(parts, self.mapping_trie, raise_on_missing)
239
240        if resolved_parts is None:
241            return None
242
243        return self.nested_get(resolved_parts, raise_on_missing=raise_on_missing)
244
245    def find_udf(self, udf: exp.Anonymous, raise_on_missing: bool = False) -> t.Optional[t.Any]:
246        """
247        Returns the return type of a given UDF.
248
249        Args:
250            udf: the target UDF expression.
251            raise_on_missing: whether to raise if the UDF is not found.
252
253        Returns:
254            The return type of the UDF, or None if not found.
255        """
256        parts = self.udf_parts(udf)
257        resolved_parts = self._find_in_trie(parts, self.udf_trie, raise_on_missing)
258
259        if resolved_parts is None:
260            return None
261
262        return nested_get(
263            self.udf_mapping,
264            *zip(resolved_parts, reversed(resolved_parts)),
265            raise_on_missing=raise_on_missing,
266        )
267
268    def nested_get(
269        self, parts: t.Sequence[str], d: t.Optional[t.Dict] = None, raise_on_missing=True
270    ) -> t.Optional[t.Any]:
271        return nested_get(
272            d or self.mapping,
273            *zip(self.supported_table_args, reversed(parts)),
274            raise_on_missing=raise_on_missing,
275        )
276
277
278class MappingSchema(AbstractMappingSchema, Schema):
279    """
280    Schema based on a nested mapping.
281
282    Args:
283        schema: Mapping in one of the following forms:
284            1. {table: {col: type}}
285            2. {db: {table: {col: type}}}
286            3. {catalog: {db: {table: {col: type}}}}
287            4. None - Tables will be added later
288        visible: Optional mapping of which columns in the schema are visible. If not provided, all columns
289            are assumed to be visible. The nesting should mirror that of the schema:
290            1. {table: set(*cols)}}
291            2. {db: {table: set(*cols)}}}
292            3. {catalog: {db: {table: set(*cols)}}}}
293        dialect: The dialect to be used for custom type mappings & parsing string arguments.
294        normalize: Whether to normalize identifier names according to the given dialect or not.
295    """
296
297    def __init__(
298        self,
299        schema: t.Optional[t.Dict] = None,
300        visible: t.Optional[t.Dict] = None,
301        dialect: DialectType = None,
302        normalize: bool = True,
303        udf_mapping: t.Optional[t.Dict] = None,
304    ) -> None:
305        self.visible = {} if visible is None else visible
306        self.normalize = normalize
307        self._dialect = Dialect.get_or_raise(dialect)
308        self._type_mapping_cache: t.Dict[str, exp.DataType] = {}
309        self._depth = 0
310        schema = {} if schema is None else schema
311        udf_mapping = {} if udf_mapping is None else udf_mapping
312
313        super().__init__(
314            self._normalize(schema) if self.normalize else schema,
315            self._normalize_udfs(udf_mapping) if self.normalize else udf_mapping,
316        )
317
318    @property
319    def dialect(self) -> Dialect:
320        """Returns the dialect for this mapping schema."""
321        return self._dialect
322
323    @classmethod
324    def from_mapping_schema(cls, mapping_schema: MappingSchema) -> MappingSchema:
325        return MappingSchema(
326            schema=mapping_schema.mapping,
327            visible=mapping_schema.visible,
328            dialect=mapping_schema.dialect,
329            normalize=mapping_schema.normalize,
330            udf_mapping=mapping_schema.udf_mapping,
331        )
332
333    def find(
334        self, table: exp.Table, raise_on_missing: bool = True, ensure_data_types: bool = False
335    ) -> t.Optional[t.Any]:
336        schema = super().find(
337            table, raise_on_missing=raise_on_missing, ensure_data_types=ensure_data_types
338        )
339        if ensure_data_types and isinstance(schema, dict):
340            schema = {
341                col: self._to_data_type(dtype) if isinstance(dtype, str) else dtype
342                for col, dtype in schema.items()
343            }
344
345        return schema
346
347    def copy(self, **kwargs) -> MappingSchema:
348        return MappingSchema(
349            **{  # type: ignore
350                "schema": self.mapping.copy(),
351                "visible": self.visible.copy(),
352                "dialect": self.dialect,
353                "normalize": self.normalize,
354                "udf_mapping": self.udf_mapping.copy(),
355                **kwargs,
356            }
357        )
358
359    def add_table(
360        self,
361        table: exp.Table | str,
362        column_mapping: t.Optional[ColumnMapping] = None,
363        dialect: DialectType = None,
364        normalize: t.Optional[bool] = None,
365        match_depth: bool = True,
366    ) -> None:
367        """
368        Register or update a table. Updates are only performed if a new column mapping is provided.
369        The added table must have the necessary number of qualifiers in its path to match the schema's nesting level.
370
371        Args:
372            table: the `Table` expression instance or string representing the table.
373            column_mapping: a column mapping that describes the structure of the table.
374            dialect: the SQL dialect that will be used to parse `table` if it's a string.
375            normalize: whether to normalize identifiers according to the dialect of interest.
376            match_depth: whether to enforce that the table must match the schema's depth or not.
377        """
378        normalized_table = self._normalize_table(table, dialect=dialect, normalize=normalize)
379
380        if match_depth and not self.empty and len(normalized_table.parts) != self.depth():
381            raise SchemaError(
382                f"Table {normalized_table.sql(dialect=self.dialect)} must match the "
383                f"schema's nesting level: {self.depth()}."
384            )
385
386        normalized_column_mapping = {
387            self._normalize_name(key, dialect=dialect, normalize=normalize): value
388            for key, value in ensure_column_mapping(column_mapping).items()
389        }
390
391        schema = self.find(normalized_table, raise_on_missing=False)
392        if schema and not normalized_column_mapping:
393            return
394
395        parts = self.table_parts(normalized_table)
396
397        nested_set(self.mapping, tuple(reversed(parts)), normalized_column_mapping)
398        new_trie([parts], self.mapping_trie)
399
400    def column_names(
401        self,
402        table: exp.Table | str,
403        only_visible: bool = False,
404        dialect: DialectType = None,
405        normalize: t.Optional[bool] = None,
406    ) -> t.List[str]:
407        normalized_table = self._normalize_table(table, dialect=dialect, normalize=normalize)
408
409        schema = self.find(normalized_table)
410        if schema is None:
411            return []
412
413        if not only_visible or not self.visible:
414            return list(schema)
415
416        visible = self.nested_get(self.table_parts(normalized_table), self.visible) or []
417        return [col for col in schema if col in visible]
418
419    def get_column_type(
420        self,
421        table: exp.Table | str,
422        column: exp.Column | str,
423        dialect: DialectType = None,
424        normalize: t.Optional[bool] = None,
425    ) -> exp.DataType:
426        normalized_table = self._normalize_table(table, dialect=dialect, normalize=normalize)
427
428        normalized_column_name = self._normalize_name(
429            column if isinstance(column, str) else column.this, dialect=dialect, normalize=normalize
430        )
431
432        table_schema = self.find(normalized_table, raise_on_missing=False)
433        if table_schema:
434            column_type = table_schema.get(normalized_column_name)
435
436            if isinstance(column_type, exp.DataType):
437                return column_type
438            elif isinstance(column_type, str):
439                return self._to_data_type(column_type, dialect=dialect)
440
441        return exp.DataType.build("unknown")
442
443    def get_udf_type(
444        self,
445        udf: exp.Anonymous | str,
446        dialect: DialectType = None,
447        normalize: t.Optional[bool] = None,
448    ) -> exp.DataType:
449        """
450        Get the return type of a UDF.
451
452        Args:
453            udf: the UDF expression or string (e.g., "db.my_func()").
454            dialect: the SQL dialect for parsing string arguments.
455            normalize: whether to normalize identifiers.
456
457        Returns:
458            The return type as a DataType, or UNKNOWN if not found.
459        """
460        parts = self._normalize_udf(udf, dialect=dialect, normalize=normalize)
461        resolved_parts = self._find_in_trie(parts, self.udf_trie, raise_on_missing=False)
462
463        if resolved_parts is None:
464            return exp.DataType.build("unknown")
465
466        udf_type = nested_get(
467            self.udf_mapping,
468            *zip(resolved_parts, reversed(resolved_parts)),
469            raise_on_missing=False,
470        )
471
472        if isinstance(udf_type, exp.DataType):
473            return udf_type
474        elif isinstance(udf_type, str):
475            return self._to_data_type(udf_type, dialect=dialect)
476
477        return exp.DataType.build("unknown")
478
479    def has_column(
480        self,
481        table: exp.Table | str,
482        column: exp.Column | str,
483        dialect: DialectType = None,
484        normalize: t.Optional[bool] = None,
485    ) -> bool:
486        normalized_table = self._normalize_table(table, dialect=dialect, normalize=normalize)
487
488        normalized_column_name = self._normalize_name(
489            column if isinstance(column, str) else column.this, dialect=dialect, normalize=normalize
490        )
491
492        table_schema = self.find(normalized_table, raise_on_missing=False)
493        return normalized_column_name in table_schema if table_schema else False
494
495    def _normalize(self, schema: t.Dict) -> t.Dict:
496        """
497        Normalizes all identifiers in the schema.
498
499        Args:
500            schema: the schema to normalize.
501
502        Returns:
503            The normalized schema mapping.
504        """
505        normalized_mapping: t.Dict = {}
506        flattened_schema = flatten_schema(schema)
507        error_msg = "Table {} must match the schema's nesting level: {}."
508
509        for keys in flattened_schema:
510            columns = nested_get(schema, *zip(keys, keys))
511
512            if not isinstance(columns, dict):
513                raise SchemaError(error_msg.format(".".join(keys[:-1]), len(flattened_schema[0])))
514            if not columns:
515                raise SchemaError(f"Table {'.'.join(keys[:-1])} must have at least one column")
516            if isinstance(first(columns.values()), dict):
517                raise SchemaError(
518                    error_msg.format(
519                        ".".join(keys + flatten_schema(columns)[0]), len(flattened_schema[0])
520                    ),
521                )
522
523            normalized_keys = [self._normalize_name(key, is_table=True) for key in keys]
524            for column_name, column_type in columns.items():
525                nested_set(
526                    normalized_mapping,
527                    normalized_keys + [self._normalize_name(column_name)],
528                    column_type,
529                )
530
531        return normalized_mapping
532
533    def _normalize_udfs(self, udfs: t.Dict) -> t.Dict:
534        """
535        Normalizes all identifiers in the UDF mapping.
536
537        Args:
538            udfs: the UDF mapping to normalize.
539
540        Returns:
541            The normalized UDF mapping.
542        """
543        normalized_mapping: t.Dict = {}
544
545        for keys in flatten_schema(udfs, depth=dict_depth(udfs)):
546            udf_type = nested_get(udfs, *zip(keys, keys))
547            normalized_keys = [self._normalize_name(key, is_table=True) for key in keys]
548            nested_set(normalized_mapping, normalized_keys, udf_type)
549
550        return normalized_mapping
551
552    def _normalize_udf(
553        self,
554        udf: exp.Anonymous | str,
555        dialect: DialectType = None,
556        normalize: t.Optional[bool] = None,
557    ) -> t.List[str]:
558        """
559        Extract and normalize UDF parts for lookup.
560
561        Args:
562            udf: the UDF expression or qualified string (e.g., "db.my_func()").
563            dialect: the SQL dialect for parsing.
564            normalize: whether to normalize identifiers.
565
566        Returns:
567            A list of normalized UDF parts (reversed for trie lookup).
568        """
569        dialect = dialect or self.dialect
570        normalize = self.normalize if normalize is None else normalize
571
572        if isinstance(udf, str):
573            parsed: exp.Expression = exp.maybe_parse(udf, dialect=dialect)
574
575            if isinstance(parsed, exp.Anonymous):
576                udf = parsed
577            elif isinstance(parsed, exp.Dot) and isinstance(parsed.expression, exp.Anonymous):
578                udf = parsed.expression
579            else:
580                raise SchemaError(f"Unable to parse UDF from: {udf!r}")
581        parts = self.udf_parts(udf)
582
583        if normalize:
584            parts = [self._normalize_name(part, dialect=dialect, is_table=True) for part in parts]
585
586        return parts
587
588    def _normalize_table(
589        self,
590        table: exp.Table | str,
591        dialect: DialectType = None,
592        normalize: t.Optional[bool] = None,
593    ) -> exp.Table:
594        dialect = dialect or self.dialect
595        normalize = self.normalize if normalize is None else normalize
596
597        normalized_table = exp.maybe_parse(table, into=exp.Table, dialect=dialect, copy=normalize)
598
599        if normalize:
600            for part in normalized_table.parts:
601                if isinstance(part, exp.Identifier):
602                    part.replace(
603                        normalize_name(part, dialect=dialect, is_table=True, normalize=normalize)
604                    )
605
606        return normalized_table
607
608    def _normalize_name(
609        self,
610        name: str | exp.Identifier,
611        dialect: DialectType = None,
612        is_table: bool = False,
613        normalize: t.Optional[bool] = None,
614    ) -> str:
615        return normalize_name(
616            name,
617            dialect=dialect or self.dialect,
618            is_table=is_table,
619            normalize=self.normalize if normalize is None else normalize,
620        ).name
621
622    def depth(self) -> int:
623        if not self.empty and not self._depth:
624            # The columns themselves are a mapping, but we don't want to include those
625            self._depth = super().depth() - 1
626        return self._depth
627
628    def _to_data_type(self, schema_type: str, dialect: DialectType = None) -> exp.DataType:
629        """
630        Convert a type represented as a string to the corresponding `sqlglot.exp.DataType` object.
631
632        Args:
633            schema_type: the type we want to convert.
634            dialect: the SQL dialect that will be used to parse `schema_type`, if needed.
635
636        Returns:
637            The resulting expression type.
638        """
639        if schema_type not in self._type_mapping_cache:
640            dialect = Dialect.get_or_raise(dialect) if dialect else self.dialect
641            udt = dialect.SUPPORTS_USER_DEFINED_TYPES
642
643            try:
644                expression = exp.DataType.build(schema_type, dialect=dialect, udt=udt)
645                expression.transform(dialect.normalize_identifier, copy=False)
646                self._type_mapping_cache[schema_type] = expression
647            except AttributeError:
648                in_dialect = f" in dialect {dialect}" if dialect else ""
649                raise SchemaError(f"Failed to build type '{schema_type}'{in_dialect}.")
650
651        return self._type_mapping_cache[schema_type]
652
653
654def normalize_name(
655    identifier: str | exp.Identifier,
656    dialect: DialectType = None,
657    is_table: bool = False,
658    normalize: t.Optional[bool] = True,
659) -> exp.Identifier:
660    if isinstance(identifier, str):
661        identifier = exp.parse_identifier(identifier, dialect=dialect)
662
663    if not normalize:
664        return identifier
665
666    # this is used for normalize_identifier, bigquery has special rules pertaining tables
667    identifier.meta["is_table"] = is_table
668    return Dialect.get_or_raise(dialect).normalize_identifier(identifier)
669
670
671def ensure_schema(schema: Schema | t.Optional[t.Dict], **kwargs: t.Any) -> Schema:
672    if isinstance(schema, Schema):
673        return schema
674
675    return MappingSchema(schema, **kwargs)
676
677
678def ensure_column_mapping(mapping: t.Optional[ColumnMapping]) -> t.Dict:
679    if mapping is None:
680        return {}
681    elif isinstance(mapping, dict):
682        return mapping
683    elif isinstance(mapping, str):
684        col_name_type_strs = [x.strip() for x in mapping.split(",")]
685        return {
686            name_type_str.split(":")[0].strip(): name_type_str.split(":")[1].strip()
687            for name_type_str in col_name_type_strs
688        }
689    elif isinstance(mapping, list):
690        return {x.strip(): None for x in mapping}
691
692    raise ValueError(f"Invalid mapping provided: {type(mapping)}")
693
694
695def flatten_schema(
696    schema: t.Dict, depth: t.Optional[int] = None, keys: t.Optional[t.List[str]] = None
697) -> t.List[t.List[str]]:
698    tables = []
699    keys = keys or []
700    depth = dict_depth(schema) - 1 if depth is None else depth
701
702    for k, v in schema.items():
703        if depth == 1 or not isinstance(v, dict):
704            tables.append(keys + [k])
705        elif depth >= 2:
706            tables.extend(flatten_schema(v, depth - 1, keys + [k]))
707
708    return tables
709
710
711def nested_get(
712    d: t.Dict, *path: t.Tuple[str, str], raise_on_missing: bool = True
713) -> t.Optional[t.Any]:
714    """
715    Get a value for a nested dictionary.
716
717    Args:
718        d: the dictionary to search.
719        *path: tuples of (name, key), where:
720            `key` is the key in the dictionary to get.
721            `name` is a string to use in the error if `key` isn't found.
722
723    Returns:
724        The value or None if it doesn't exist.
725    """
726    for name, key in path:
727        d = d.get(key)  # type: ignore
728        if d is None:
729            if raise_on_missing:
730                name = "table" if name == "this" else name
731                raise ValueError(f"Unknown {name}: {key}")
732            return None
733
734    return d
735
736
737def nested_set(d: t.Dict, keys: t.Sequence[str], value: t.Any) -> t.Dict:
738    """
739    In-place set a value for a nested dictionary
740
741    Example:
742        >>> nested_set({}, ["top_key", "second_key"], "value")
743        {'top_key': {'second_key': 'value'}}
744
745        >>> nested_set({"top_key": {"third_key": "third_value"}}, ["top_key", "second_key"], "value")
746        {'top_key': {'third_key': 'third_value', 'second_key': 'value'}}
747
748    Args:
749        d: dictionary to update.
750        keys: the keys that makeup the path to `value`.
751        value: the value to set in the dictionary for the given key path.
752
753    Returns:
754        The (possibly) updated dictionary.
755    """
756    if not keys:
757        return d
758
759    if len(keys) == 1:
760        d[keys[0]] = value
761        return d
762
763    subd = d
764    for key in keys[:-1]:
765        if key not in subd:
766            subd = subd.setdefault(key, {})
767        else:
768            subd = subd[key]
769
770    subd[keys[-1]] = value
771    return d