(mcls, name, bases, dct)
| 48 | } |
| 49 | |
| 50 | def __new__(mcls, name, bases, dct): |
| 51 | cls = super().__new__(mcls, name, bases, dct) |
| 52 | if cls.__module__ == mcls.__module__ and name == 'PostgresMessage': |
| 53 | for f in mcls._field_map.values(): |
| 54 | setattr(cls, f, None) |
| 55 | |
| 56 | if _is_asyncpg_class(cls): |
| 57 | mod = sys.modules[cls.__module__] |
| 58 | if hasattr(mod, name): |
| 59 | raise RuntimeError('exception class redefinition: {}'.format( |
| 60 | name)) |
| 61 | |
| 62 | code = dct.get('sqlstate') |
| 63 | if code is not None: |
| 64 | existing = mcls._message_map.get(code) |
| 65 | if existing is not None: |
| 66 | raise TypeError('{} has duplicate SQLSTATE code, which is' |
| 67 | 'already defined by {}'.format( |
| 68 | name, existing.__name__)) |
| 69 | mcls._message_map[code] = cls |
| 70 | |
| 71 | return cls |
| 72 | |
| 73 | @classmethod |
| 74 | def get_message_class_for_sqlstate(mcls, code): |
nothing calls this directly
no test coverage detected