| 673 | return util.PopulateDict(self._kw_reg_for_dialect_cls) |
| 674 | |
| 675 | def _validate_dialect_kwargs(self, kwargs: Dict[str, Any]) -> None: |
| 676 | # validate remaining kwargs that they all specify DB prefixes |
| 677 | |
| 678 | if not kwargs: |
| 679 | return |
| 680 | |
| 681 | for k in kwargs: |
| 682 | m = re.match("^(.+?)_(.+)$", k) |
| 683 | if not m: |
| 684 | raise TypeError( |
| 685 | "Additional arguments should be " |
| 686 | "named <dialectname>_<argument>, got '%s'" % k |
| 687 | ) |
| 688 | dialect_name, arg_name = m.group(1, 2) |
| 689 | |
| 690 | try: |
| 691 | construct_arg_dictionary = self.dialect_options[dialect_name] |
| 692 | except exc.NoSuchModuleError: |
| 693 | util.warn( |
| 694 | "Can't validate argument %r; can't " |
| 695 | "locate any SQLAlchemy dialect named %r" |
| 696 | % (k, dialect_name) |
| 697 | ) |
| 698 | self.dialect_options[dialect_name] = d = _DialectArgDict() |
| 699 | d._defaults.update({"*": None}) |
| 700 | d._non_defaults[arg_name] = kwargs[k] |
| 701 | else: |
| 702 | if ( |
| 703 | "*" not in construct_arg_dictionary |
| 704 | and arg_name not in construct_arg_dictionary |
| 705 | ): |
| 706 | raise exc.ArgumentError( |
| 707 | "Argument %r is not accepted by " |
| 708 | "dialect %r on behalf of %r" |
| 709 | % (k, dialect_name, self.__class__) |
| 710 | ) |
| 711 | else: |
| 712 | construct_arg_dictionary[arg_name] = kwargs[k] |
| 713 | |
| 714 | |
| 715 | class CompileState: |