|
|
@ -46,7 +46,7 @@ class Db(object): |
|
|
|
information['size'] = self._runsql("SELECT pg_size_pretty(pg_database_size('"+self._name+"'));").strip() |
|
|
|
information['encoding'], information['collation'], \ |
|
|
|
information['ctype'] = self._runsql("""SELECT pg_encoding_to_char(encoding), |
|
|
|
datcollate, datctype FROM pg_database WHERE datname='"""+self._name+"';").strip().split('|') |
|
|
|
datcollate, datctype FROM pg_database WHERE datname='"""+self._name+"';").split('|') |
|
|
|
return information |
|
|
|
|
|
|
|
@property |
|
|
@ -57,14 +57,14 @@ class Db(object): |
|
|
|
|
|
|
|
def user_exists(self, user): |
|
|
|
sql = "SELECT rolname FROM pg_authid WHERE rolname = '"+user+"';" |
|
|
|
u = self._runsql(sql).strip() |
|
|
|
u = self._runsql(sql) |
|
|
|
if (u == ""): |
|
|
|
return False |
|
|
|
return True |
|
|
|
|
|
|
|
def db_exists(self, xdb): |
|
|
|
sql = "SELECT datname FROM pg_database WHERE datname = '"+xdb+"';" |
|
|
|
d = self._runsql(sql).strip() |
|
|
|
d = self._runsql(sql) |
|
|
|
if (d == ""): |
|
|
|
return False |
|
|
|
return True |
|
|
@ -112,8 +112,8 @@ class Db(object): |
|
|
|
return usrlist |
|
|
|
|
|
|
|
def _test(): |
|
|
|
#test = Db(u'postgres') |
|
|
|
#print test.info['encoding'], test.info['collation'], test.info['ctype'] |
|
|
|
test = Db(u'postgres') |
|
|
|
print test.info['encoding'], test.info['collation'], test.info['ctype'] |
|
|
|
#print test.owner |
|
|
|
#print test.connections |
|
|
|
#print "User aaa is ",test.user_exists("aaa") |
|
|
@ -130,9 +130,9 @@ def _test(): |
|
|
|
#print test2.owner |
|
|
|
#test2.owner = u'karasz' |
|
|
|
#print test2.owner |
|
|
|
test = Db(u'aaa') |
|
|
|
test.create(u'postgres', u'en_US.UTF-8', u'en_US.UTF-8') |
|
|
|
print test.dump('/tmp/aaa.gz',u'p') |
|
|
|
#test = Db(u'aaa') |
|
|
|
#test.create(u'postgres', u'en_US.UTF-8', u'en_US.UTF-8') |
|
|
|
#print test.dump('/tmp/aaa.gz',u'p') |
|
|
|
#test.delete() |
|
|
|
if __name__ == '__main__': |
|
|
|
_test() |
|
|
|