如何解决手动和通过 Postgres 中的代码运行“REVOKE CREATE ON schema public FROM public”时获得不同的输出
在 postgres 中，我使用 superuser_name 创建了“test”数据库，然后我创建了新用户“app_user”。然后我从任何公共模式的创建表中撤销 app_user。然后我将测试数据库的所有权限都授予了 app_user。所以这种方式 app_user 只能在测试数据库中创建表。步骤：
- psql -h localhost -p 5432 -U superuser_name
- CREATE DATABASE test WITH OWNER superuser_name ENCODING 'UTF8'
- CREATE USER app_user WITH PASSWORD 'app_password'
- REVOKE CREATE ON schema public FROM public
- GRANT ALL PRIVILEGES ON DATABASE test TO app_user
- now quit and login back as app_user
- psql -h localhost -p 5432 -U app_user -d test
测试=> \dn+
List of schemas
Name | Owner | Access privileges | Description
--------+----------+----------------------+------------------------
public | postgres | postgres=UC/postgres+| standard public schema
| | =UC/postgres |
(1 row)
- 按照上述步骤，app_user 只能在测试数据库中创建表，而不允许在任何其他数据库中创建表。这正是我所期待的。但是当我通过代码运行上述步骤时，我没有得到相同的行为。这意味着，新创建的 app_user 无权在新创建的测试数据库上创建表。以及 dn+ 命令的输出如下。
测试=> \dn+
List of schemas
Name | Owner | Access privileges | Description
--------+----------+----------------------+------------------------
public | postgres | postgres=UC/postgres+| standard public schema
| | =U/postgres |
(1 row)
Python 代码，用于创建数据库、用户、授予和撤消我正在运行的权限。
from collections import namedtuple
from sqlalchemy.orm import Session
from sqlalchemy import create_engine
from contextlib import contextmanager
from urllib.parse import quote_plus
from dotenv import load_dotenv
import time,os
import logging
log = logging.getLogger(__name__)
load_dotenv()
SLEEP_TIME_BEFORE_CREATING_TABLES = 60
PG_ENGINE_URL = 'postgresql://{user}:{password}@{host}:{port}/{dbname}'
DbConfig = namedtuple('DbConfig','host port user password dbname')
def postgres_config():
"""
Return postgres connection config.
"""
pg_cfg = {
'host': os.environ.get('HOST','localhost'),'port': os.environ.get('PORT',5432),'superuser_name': os.environ.get('SUPERUSER_NAME','postgres'),'superuser_password': os.environ.get('SUPERUSER_PASSWORD','app_user': os.environ.get('APP_USERNAME','visualrf'),'app_password': os.environ.get('APP_PASSWORD','visualrf123'),}
return pg_cfg
def engine_url(cfg):
url = PG_ENGINE_URL.format(host=cfg.host,port=cfg.port,dbname=cfg.dbname,user=cfg.user,password=quote_plus(cfg.password))
return url
def engine(cfg,serialize=False):
"""
Return postgres engine url to connect with postgress databse.
"""
if serialize:
return create_engine(engine_url(cfg),isolation_level="SERIALIZABLE",client_encoding='utf8')
else:
return create_engine(engine_url(cfg),client_encoding='utf8')
def session(cfg):
"""
Return sqlalchemy session to database using QueuePool.
"""
return Session(bind=engine(cfg),expire_on_commit=False)
def anon_session(host,port,user,password):
"""
Make a session to template1 db which is always present in postgres.
Useful to execute SQL before a particular db is created.
"""
return session(DbConfig(host,password,'template1'))
@contextmanager
def terminating_sn(sn_or_cfg):
""" A contextlib which closes session and db connections after use. """
sn = session(sn_or_cfg) if isinstance(sn_or_cfg,DbConfig) else sn_or_cfg
try:
yield sn
finally:
sn.close()
sn.bind.dispose()
def db_exists(host,dbname):
with terminating_sn(anon_session(host,password)) as sn:
r = sn.execute("SELECT 1 FROM pg_catalog.pg_database d WHERE d.datname='%s'" % dbname).fetchone()
return True if r else False
def _db_exists(sn,dbname):
""" True if DB exists """
r = sn.execute("SELECT 1 FROM pg_catalog.pg_database d WHERE d.datname='%s'" % dbname).fetchone()
return True if r else False
def _user_exists(sn,app_user):
""" True if user/rolname exists """
r = sn.execute("SELECT 1 FROM pg_user WHERE pg_user.usename='%s'" % (app_user))
return True if r.rowcount!=0 else False
def create_db(dbname,dbconf,owner=None):
""" Pick a DB host and create a database in it """
owner = owner or dbconf['superuser_name']
with terminating_sn(anon_session(dbconf['host'],dbconf['port'],dbconf['superuser_name'],dbconf['superuser_password'])) as sn: # connecting to postgres session
sn.connection().connection.set_isolation_level(0)
if _db_exists(sn,dbname):
return False
sn.execute("CREATE DATABASE %s WITH OWNER %s ENCODING 'UTF8'" % (dbname,owner)) # create database
log.info('Provisioned db:%s for host:%s',dbname,dbconf['host'])
return True
def create_user(dbname,dbconf):
""" Create app user for <dbname> and grant att read write permition """
if db_exists(dbconf['host'],dbconf['superuser_password'],dbname): # create anon session to check db exist.
with terminating_sn(DbConfig(dbconf['host'],dbname)) as sn: # if db exist then make connection with mentioned dbname.
sn.connection().connection.set_isolation_level(0)
if not _user_exists(sn,dbconf['app_user']): # check if user already exist,if not then create one
sn.execute("CREATE USER %s WITH PASSWORD '%s'" % (dbconf['app_user'],dbconf['app_password']))
log.info("Created user:%s for db:%s on host:%s",dbconf['app_user'],dbconf['host'])
else:
log.info("User '%s' already exist for db:%s on host:%s",dbconf['host'])
#Grant all privileges to app_user on the database:
sn.execute("REVOKE CREATE ON schema public FROM public")
sn.execute("GRANT ALL PRIVILEGES ON DATABASE %s TO %s" % (dbname,dbconf['app_user']))
return True
else:
log.error("Can not create user:%s because db:%s for host:%s do not exist",dbconf['host'])
return False
def create_table(dbname,sqla_metadata):
""" create tables for <dbname> database using created app user """
if db_exists(dbconf['host'],dbconf['app_password'],dbname) and sqla_metadata: # if database exist and we have metabadat to create tables
with terminating_sn(DbConfig(dbconf['host'],dbname)) as sn:
sqla_metadata.create_all(bind=sn.bind) # create tables
else:
log.error('Can not create tables because db:%s for host:%s do not exist',dbconf['host'])
return False
log.info('Created tables for db:%s on host:%s',dbconf['host'])
return True
所以我无法弄清楚为什么我有这种不同的行为。为什么 app_user 通过 python 脚本无法在测试数据库中创建表？
版权声明：本文内容由互联网用户自发贡献，该文观点与技术仅代表作者本人。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容， 请发送邮件至 dio@foxmail.com 举报，一经查实，本站将立刻删除。
