Configure a handler from a dictionary.
(self, config)
| 766 | return handler |
| 767 | |
| 768 | def configure_handler(self, config): |
| 769 | """Configure a handler from a dictionary.""" |
| 770 | config_copy = dict(config) # for restoring in case of error |
| 771 | formatter = config.pop('formatter', None) |
| 772 | if formatter: |
| 773 | try: |
| 774 | formatter = self.config['formatters'][formatter] |
| 775 | except Exception as e: |
| 776 | raise ValueError('Unable to set formatter ' |
| 777 | '%r' % formatter) from e |
| 778 | level = config.pop('level', None) |
| 779 | filters = config.pop('filters', None) |
| 780 | if '()' in config: |
| 781 | c = config.pop('()') |
| 782 | if not callable(c): |
| 783 | c = self.resolve(c) |
| 784 | factory = c |
| 785 | else: |
| 786 | cname = config.pop('class') |
| 787 | if callable(cname): |
| 788 | klass = cname |
| 789 | else: |
| 790 | klass = self.resolve(cname) |
| 791 | if issubclass(klass, logging.handlers.MemoryHandler): |
| 792 | if 'flushLevel' in config: |
| 793 | config['flushLevel'] = logging._checkLevel(config['flushLevel']) |
| 794 | if 'target' in config: |
| 795 | # Special case for handler which refers to another handler |
| 796 | try: |
| 797 | tn = config['target'] |
| 798 | th = self.config['handlers'][tn] |
| 799 | if not isinstance(th, logging.Handler): |
| 800 | config.update(config_copy) # restore for deferred cfg |
| 801 | raise TypeError('target not configured yet') |
| 802 | config['target'] = th |
| 803 | except Exception as e: |
| 804 | raise ValueError('Unable to set target handler %r' % tn) from e |
| 805 | elif issubclass(klass, logging.handlers.QueueHandler): |
| 806 | # Another special case for handler which refers to other handlers |
| 807 | # if 'handlers' not in config: |
| 808 | # raise ValueError('No handlers specified for a QueueHandler') |
| 809 | if 'queue' in config: |
| 810 | qspec = config['queue'] |
| 811 | |
| 812 | if isinstance(qspec, str): |
| 813 | q = self.resolve(qspec) |
| 814 | if not callable(q): |
| 815 | raise TypeError('Invalid queue specifier %r' % qspec) |
| 816 | config['queue'] = q() |
| 817 | elif isinstance(qspec, dict): |
| 818 | if '()' not in qspec: |
| 819 | raise TypeError('Invalid queue specifier %r' % qspec) |
| 820 | config['queue'] = self.configure_custom(dict(qspec)) |
| 821 | elif not _is_queue_like_object(qspec): |
| 822 | raise TypeError('Invalid queue specifier %r' % qspec) |
| 823 | |
| 824 | if 'listener' in config: |
| 825 | lspec = config['listener'] |
no test coverage detected