1 # Copyright (C) 2012 Nippon Telegraph and Telephone Corporation.
2 # Copyright (C) 2012 Isaku Yamahata <yamahata at private email ne jp>
4 # Licensed under the Apache License, Version 2.0 (the "License");
5 # you may not use this file except in compliance with the License.
6 # You may obtain a copy of the License at
8 # http://www.apache.org/licenses/LICENSE-2.0
10 # Unless required by applicable law or agreed to in writing, software
11 # distributed under the License is distributed on an "AS IS" BASIS,
12 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
14 # See the License for the specific language governing permissions and
15 # limitations under the License.
20 from ovs import jsonrpc
21 from ovs import stream
22 from ovs import util as ovs_util
23 from ovs.db import schema
25 LOG = logging.getLogger(__name__)
28 class DBClient(object):
29 def __init__(self, remote):
30 super(DBClient, self).__init__()
33 def run_command(self, args):
35 'list-dbs': self._list_dbs,
36 'get-schema': self._get_schema,
37 'get-schema-version': self._get_schema_version,
38 'list-tables': self._list_tables,
39 'list-columns': self._list_columns,
40 'transact': self._transact,
41 'monitor': self._monitor,
48 error, stream_ = stream.Stream.open_block(
49 stream.Stream.open(self.remote))
51 raise RuntimeError('can not open socket to %s: %s' %
52 (self.remote, os.strerror(error)))
53 rpc = jsonrpc.Connection(stream_)
55 ret = _COMMANDS[command](rpc, *args)
56 LOG.info('ret %s', ret)
59 def _check_txn(self, error, reply):
61 ovs_util.ovs_fatal(error, os.strerror(error))
63 ovs_util.ovs_fatal(reply.error, 'error %s' % reply.error)
65 def _fetch_dbs(self, rpc):
66 request = jsonrpc.Message.create_request('list_dbs', [])
67 error, reply = rpc.transact_block(request)
68 self._check_txn(error, reply)
71 for name in reply.result:
76 def _fetch_schema_json(self, rpc, database):
77 request = jsonrpc.Message.create_request('get_schema', [database])
78 error, reply = rpc.transact_block(request)
79 self._check_txn(error, reply)
82 def _fetch_schema(self, rpc, database):
83 return schema.DbSchema.from_json(self._fetch_schema_json(rpc,
87 def _list_dbs(self, rpc, *_):
88 return self._fetch_dbs(rpc)
90 def _get_schema(self, rpc, *args):
92 return self._fetch_schema(rpc, database).to_json()
94 def _get_schema_version(self, rpc, *_args):
96 schema_ = self._fetch_schema(rpc, database)
97 return schema_.version
99 def _list_tables(self, rpc, *args):
101 schema_ = self._fetch_schema(rpc, database)
102 return [table.to_json() for table in schema_.tables.values()]
104 def _list_columns(self, rpc, *args):
110 schema_ = self._fetch_schema(rpc, database)
111 if table_name is None:
112 tables = [table for table in schema_.tables.values()]
114 tables = [table for table in schema_.tables.values()
115 if table.name == table_name]
119 columns.extend(table.columns.values())
120 return [column.to_json() for column in columns]
122 def _transact(self, rpc, *args):
123 raise NotImplementedError()
125 def _monitor(self, rpc, *args):
126 raise NotImplementedError()
128 def _dump(self, rpc, *args):
129 raise NotImplementedError()