Thank you very much for your hints. They were very helpful!
Am 09.05.13 16:16, schrieb Michael Bayer:
>
> On May 9, 2013, at 6:32 AM, Wolfgang Meiners <wolfgangmeiner...@web.de> wrote:
>
>>
>> Now i want to run everything inside the schema 'test'. Two questions arises:
>> 1) How can i create schema test automatically when it does not exist?
>> This should be somehow related to
>> Base.metadata.create_all(engine)
>
> you'd use a DDL("CREATE SCHEMA <foo>") that's attached to the "before_create"
> event of the MetaData object.
that seems to be exactly what i was looking for. When i tried it out, i
run into an error because SCHEMA 'test' existed. Maybe the simplest
solution to this is
DDL('''DROP SCHEMA {s} CASCADING IF EXISTS;
CREATE SCHEMA {s};'''.format(s=Base.metadata.schema))
I came up with a slighly different solution, see below.
>
>>
>> 2) If i create schema test by hand and uncomment the line
>> #Base.metadata.schema='test'
>> i get an errormessage because the DDL does not use schema test:
>> sqlalchemy.exc.ProgrammingError: (ProgrammingError) relation "valid"
>> does not exist
>
> you'd need to refer to the table "valid" as "myschema.valid" within the
> definition of your stored procedure.
and this can be achieved with a
"... {s}valid ...".\
format(s=Base.metadata.schema + '.' if Base.metadata.schema else '')
For the interested, here is a solution that works:
=====================================================================
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
'''
Created on 09.05.2013
Database interface for postgresql 9.1
Uses sqlalchemy 0.7.10, psycopg2 2.4.6, postgresql 9.1
@author: wolfgang
'''
from sqlalchemy import Integer,Column, create_engine, DDL, event,
select, text
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm.session import sessionmaker
from locale import setlocale, LC_ALL
Base=declarative_base()
Base.metadata.schema='test'
class Invalid(Base):
__tablename__ = 'invalid'
number = Column(Integer, primary_key=True)
def __init__(self, number):
self.number = number
class Valid(Base):
__tablename__ = 'valid'
number = Column(Integer, primary_key=True)
def __init__(self, number):
self.number = number
trigger_ddl = DDL('''
CREATE OR REPLACE FUNCTION {s}validate_number()
RETURNS TRIGGER AS $$
BEGIN
IF EXISTS(
SELECT * FROM {s}invalid WHERE number = NEW.number
)
THEN
RAISE EXCEPTION '%% is invalid', NEW.number;
RETURN NULL;
END IF;
RETURN NEW;
END;
$$ LANGUAGE plpgsql;
CREATE TRIGGER validate_number
BEFORE INSERT OR UPDATE ON {s}valid
FOR EACH ROW EXECUTE PROCEDURE {s}validate_number();
'''.format(s=Base.metadata.schema + '.' if Base.metadata.schema else
''))
setlocale(LC_ALL,'')
url = 'postgresql+psycopg2://{u}:{pw}@{h}:{p}/{db}'.format(
u='tester', pw='mypassword', h='localhost', p='5432', db='tester')
engine = create_engine(url, echo=True)
event.listen(Valid.__table__, 'after_create', Valid.trigger_ddl)
event.listen(Valid.__table__, 'after_drop',
DDL('DROP FUNCTION {s}validate_number() CASCADE;'.\
format(s = Base.metadata.schema + '.' if\
Base.metadata.schema else '')))
Base.metadata.drop_all(engine)
myschema = Base.metadata.schema
if myschema is not None:
t = text("SELECT schema_name FROM information_schema.schemata WHERE
schema_name = '{s}'".format(s=myschema))
r = engine.connect().execute(t).fetchall()
if not r:
ddl_before = DDL('CREATE SCHEMA {s}'.format(s=myschema))
event.listen(Base.metadata, 'before_create', ddl_before)
ddl_after = DDL('DROP SCHEMA {s}'.format(s=myschema))
event.listen(Base.metadata, 'after_drop', ddl_after)
Base.metadata.create_all(engine)
Session=sessionmaker(engine)
session = Session()
session.add(Invalid(3))
session.commit()
#works
session.add(Valid(1))
session.add(Valid(2))
session.commit()
#raises an error
session.add(Valid(3))
try:
session.commit()
except:
session.rollback()
qv = session.query(Valid)
for v in qv: print('number {v} is valid'.format(v=v.number))
session.close()
Base.metadata.drop_all(engine)
=====================================================================
--
You received this message because you are subscribed to the Google Groups
"sqlalchemy" group.
To unsubscribe from this group and stop receiving emails from it, send an email
to sqlalchemy+unsubscr...@googlegroups.com.
To post to this group, send email to sqlalchemy@googlegroups.com.
Visit this group at http://groups.google.com/group/sqlalchemy?hl=en.
For more options, visit https://groups.google.com/groups/opt_out.
