Ich versuche, einen Datenbanknamen dynamisch zu meinen Verbindungsinformationen hinzuzufügen (dies ist ein Wörterbuch mit Schlüsselwerten für die Verbindung zu beispielsweise psycopg2
). Dafür benutze ich eine Eigenschaft. Wie ich mir vorstelle, dass es in meinem Kopf funktioniert, ist, dass es die Informationen an den Objekt-Ersteller übergibt und dann das 'database'
Attribut in meinem canvas_conn
Wörterbuch auf database_name
setzt. Dann habe ich eine @property
Methode für hinzugefügt die database_name
für den Fall, dass es jemals geändert wird, um die canvas_conn
zu aktualisieren. Mein Komponententest schlägt jedoch mit KeyError
für "Datenbank" fehl.@property bei der Objekterstellung nicht ordnungsgemäß aktualisiert?
Hier ist mein Code:
class Shard(mu.PrettyStr):
def __init__(self,
database_name,
schema_name,
cluster_id,
shard_id,
historical_connect=False,
canvas_conn=config3.CANVAS_INFO,
analytics_conn=config3.REDSHIFT_POSTGRES_INFO_PROD):
"""
@type database_name: str
:param server_id:
@type schema_name: str
:param name:
@type cluster_id: str
:param cluster_id:
@type shard_id: str
:param shard_id:
@type historical_connect: bool
:param historical_connect:
@type canvas_conn: dict
:param canvas_conn:
@type analytics_conn: dict
:param analytics_conn:
"""
self._database_name = database_name
self.schema_name = schema_name
self.cluster_id = cluster_id
self.shard_id = shard_id
self.historical_connect = historical_connect
self._canvas_conn = canvas_conn
self.analytics_conn = analytics_conn
@property
def database_name(self):
return self._database_name
@database_name.setter
def database_name(self, name):
self._database_name = name
# force the new database name into the conn info
self._canvas_conn = self._canvas_conn
@property
def canvas_conn(self):
return self._canvas_conn
@canvas_conn.setter
def canvas_conn(self, database_info):
database_info['database'] = self._database_name
self._canvas_conn = database_info
Und hier ist meine Unit-Test:
def test_descriptor_correctly_sets_database_attribute(self):
test_data = ('server_id', 'db_name', 'cluster_id', 'shard_id', True)
test_shard = cu.Shard(*test_data)
self.assertEqual(
test_shard.canvas_conn['database'],
test_shard.database_name
)