# -*- coding: utf-8 -*- # !/usr/bin/env python from mongoengine import DynamicDocument, StringField, IntField, BooleanField, DoesNotExist from typing import TYPE_CHECKING if TYPE_CHECKING: from pymongo.collection import Collection class ArchivedDatabaseConfig(DynamicDocument): alias = StringField(verbose_name = "连接别名", required = True, unique = True) name = StringField(verbose_name = "数据库名称", required = True) host = StringField(verbose_name = "连接IP", required = True) port = IntField(verbose_name = "端口", default = 27017) username = StringField(verbose_name = "用户名", default = "admin") password = StringField(verbose_name = "密码", required = True) authentication_source = StringField(verbose_name = "认证数据库名称", default = "admin") active = BooleanField(verbose_name = "是否启用该数据库连接", default = True) meta = { "collection": "ArchivedDatabaseConfig", "db_alias": "default" } @classmethod def get_all_connection(cls): connInfo = cls.objects.filter(active = True) return map(lambda x: x.to_dict(), connInfo) def to_dict(self): return { "alias": self.alias, "name": self.name, "host": self.host, "port": self.port, "username": self.username, "password": self.password, "authentication_source": self.authentication_source } class ArchivedModelProxyConfig(DynamicDocument): model = StringField(verbose_name = u'ORM名称', required = True) startDay = StringField(verbose_name = u'热点数据开始日期', required = True) meta = { "collection": "ArchivedModelProxyConfig", "db_alias": "default" } @classmethod def get_model_hot_data_start_day(cls, model_name): try: item = cls.objects.get(model = model_name) return item.startDay except DoesNotExist as e: # 默认 的现网数据库数据保存的月数 if model_name == 'default': return '2017-07-01' else: return cls.get_model_hot_data_start_day('default') class MongoTempTable(DynamicDocument): _origin_meta = {'db_alias': 'logdata'} @classmethod def get_collection(cls): # type: ()->Collection return cls._get_collection()