In postgres, I created "test" database with superuser_name and then I created new user "app_user". Then I revoke app_user from create table for any public schema. And then I gave all permission on test database to app_user. So this way app_user can create tables in only test database. The steps:

- psql -h localhost -p 5432 -U superuser_name
- CREATE USER app_user WITH PASSWORD 'app_password'
- REVOKE CREATE ON schema public FROM public
- now quit and login back as app_user
- psql -h localhost -p 5432 -U app_user -d test

test=> \dn+

                             List of schemas
      Name  |  Owner   |  Access privileges   |      Description       
     public | postgres | postgres=UC/postgres+| standard public schema
            |          | =UC/postgres         | 
    (1 row)
  • by following above steps, app_user is able to create table only in test database and not allow to create table in any other databases. That is what I was expecting. But When I run above steps via code then I am not getting same behaviors. Meaning, newly created app_user do not have permission to create table on newly created test database. And output for dn+ command as below.

test=> \dn+

                        List of schemas
     Name  |  Owner   |  Access privileges   |      Description       
    public | postgres | postgres=UC/postgres+| standard public schema
           |          | =U/postgres         | 
   (1 row)

Python code, to create database, user, give and revoke permission, that I am running.

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__)

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','postgres'),
        '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",
        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, port, user, password, 'template1'))

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
        yield sn

def db_exists(host, port, user, password, dbname):
    with terminating_sn(anon_session(host, port, user, 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
        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['port'], dbconf['superuser_name'], dbconf['superuser_password'], dbname): # create anon session to check db exist.
        with terminating_sn(DbConfig(dbconf['host'], dbconf['port'], dbconf['superuser_name'], dbconf['superuser_password'], dbname)) as sn: # if db exist then make connection with mentioned dbname.
            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'], dbname, dbconf['host'])
                log.info("User '%s' already exist for db:%s on host:%s", dbconf['app_user'], dbname, 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
        log.error("Can not create user:%s because db:%s for host:%s do not exist", dbconf['app_user'], dbname, dbconf['host'])
        return False

def create_table(dbname, dbconf, sqla_metadata):
    """ create tables for <dbname> database using created app user """
    if db_exists(dbconf['host'], dbconf['port'], dbconf['app_user'], dbconf['app_password'], dbname) and sqla_metadata:  # if database exist and we have metabadat to create tables 
        with terminating_sn(DbConfig(dbconf['host'], dbconf['port'], dbconf['app_user'], dbconf['app_password'], dbname)) as sn:
            sqla_metadata.create_all(bind=sn.bind)  # create tables
        log.error('Can not create tables because db:%s for host:%s do not exist', dbname, dbconf['host'])
        return False
    log.info('Created tables for db:%s on host:%s', dbname, dbconf['host'])
    return True

And above python script will execute in this order:

created_db = pg.create_db(dbname, pg_cfg)
created_user = pg.create_user(dbname, pg_cfg) # creating app user and 
    grant all privileges for <dbname> database
created_table = pg.create_table(dbname, pg_cfg, AppBase.metadata)

So I am not able to figured out Why I am having this different behavior. Why app_user via python script not able to create table in test database?

Also, My requirements is, I just want to create new user and new db. then grant all privileges to new user for new db. and prevent new user to create tables in any other database except new db. what is the straightforward way to achieve this?


1 Answer 1


The statement


affects the database to which you are currently connected.

Since the script you are showing in the beginning does not contain a connect to the newly created database, the statement will affect the database to which you were connected when you ran CREATE DATABASE rather than the newly created database.

After creating the database, add the following to your psql script:

\c test

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.

Not the answer you're looking for? Browse other questions tagged or ask your own question.