This repository has been archived by the owner on Mar 24, 2021. It is now read-only.
-
Notifications
You must be signed in to change notification settings - Fork 11
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
wip: Implement PostgresStorageEngine
Not passing the tests yet...
- Loading branch information
1 parent
79d7e6a
commit 05804ce
Showing
4 changed files
with
112 additions
and
60 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,52 +1,93 @@ | ||
| class PostgresStorageEngine(object): | ||
| import psycopg2 | ||
| import json | ||
| from datetime import date, datetime | ||
| from .. import timeutils | ||
|
|
||
| def __init__(self): | ||
| pass | ||
| def json_serial(obj): | ||
| if isinstance(obj, (datetime, date)): | ||
| return obj.isoformat() | ||
| raise TypeError ("Type %s not serializable" % type(obj)) | ||
|
|
||
| def alive(self): | ||
| # TODO | ||
| pass | ||
| class PostgresStorageEngine(object): | ||
|
|
||
| def __init__(self, datatbase_url): | ||
| self.connection = psycopg2.connect(datatbase_url) | ||
|
|
||
| def data_set_exists(self, data_set_id): | ||
| # TODO | ||
| pass | ||
| # This is slightly different to the mongo implementation | ||
| # in that it will return False if `create_data_set` has | ||
| # been called, but no records have been saved. | ||
| with self.connection.cursor() as psql_cursor: | ||
| psql_cursor.execute(""" | ||
| SELECT 1 FROM mongo | ||
| WHERE collection=%(collection)s | ||
| LIMIT 1 | ||
| """, | ||
| {'collection': data_set_id} | ||
| ) | ||
| return psql_cursor.fetchone() is not None | ||
|
|
||
| def create_data_set(self, data_set_id, size): | ||
| # TODO | ||
| pass | ||
|
|
||
| def delete_data_set(self, data_set_id): | ||
| # TODO | ||
| pass | ||
| with self.connection.cursor() as psql_cursor: | ||
| psql_cursor.execute( | ||
| "DELETE FROM mongo WHERE collection=%(collection)s", | ||
| {'collection': data_set_id} | ||
| ) | ||
| self.connection.commit() | ||
|
|
||
| def get_last_updated(self, data_set_id): | ||
| # TODO | ||
| # TODO - this requires a bit of sorting on the JSON column | ||
| pass | ||
|
|
||
| def batch_last_updated(self, data_sets): | ||
| # TODO | ||
| # TODO - this requires quite a complex query | ||
| pass | ||
|
|
||
| def empty_data_set(self, data_set_id): | ||
| # TODO | ||
| pass | ||
| delete_data_set(data_set_id) | ||
|
|
||
| def save_record(self, data_set_id, record): | ||
| # TODO | ||
| pass | ||
| self.update_record(data_set_id, record['_id'], record) | ||
|
|
||
| def find_record(self, data_set_id, record_id): | ||
| # TODO | ||
| pass | ||
| with self.connection.cursor() as psql_cursor: | ||
| psql_cursor.execute(""" | ||
| SELECT record FROM mongo | ||
| WHERE id=%(id)s | ||
| """, | ||
| {'id': record_id} | ||
| ) | ||
| (record,) = psql_cursor.fetchone() | ||
| return record | ||
|
|
||
| def update_record(self, data_set_id, record_id, record): | ||
| # TODO | ||
| pass | ||
| record['_updated_at'] = timeutils.now() | ||
| record_id = data_set_id + ':' + record_id | ||
| with self.connection.cursor() as psql_cursor: | ||
| psql_cursor.execute(""" | ||
| INSERT INTO mongo (id, collection, record) | ||
| VALUES (%(id)s, %(collection)s, %(record)s) | ||
| ON CONFLICT (id) DO UPDATE SET record=%(record)s""", | ||
| { | ||
| 'id': record_id, | ||
| 'collection': data_set_id, | ||
| 'record': json.dumps(record, default=json_serial) | ||
| } | ||
|
|
||
| ) | ||
| self.connection.commit() | ||
|
|
||
| def delete_record(self, data_set_id, record_id): | ||
| # TODO | ||
| pass | ||
| with self.connection.cursor() as psql_cursor: | ||
| psql_cursor.execute( | ||
| "DELETE FROM mongo WHERE id=%(id)s", | ||
| {'id': record_id} | ||
| ) | ||
| self.connection.commit() | ||
|
|
||
| def execute_query(self, data_set_id, query): | ||
| # TODO | ||
| pass | ||
| return [] |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -20,3 +20,4 @@ logstash_formatter==0.5.7 | |
| Werkzeug==0.12.2 | ||
| redis==2.10.6 | ||
| flower==0.9.2 | ||
| psycopg2==2.7.5 | ||
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,10 @@ | ||
| from backdrop.core.storage.postgres import PostgresStorageEngine | ||
|
|
||
| from .test_storage import BaseStorageTest | ||
|
|
||
| class TestPostgresStorageEngine(BaseStorageTest): | ||
| def setup(self): | ||
| self.engine = PostgresStorageEngine('postgres://postgres:mysecretpassword@localhost:5432') | ||
|
|
||
| def teardown(self): | ||
| self.engine.delete_data_set('foo_bar') |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters