do me several huge favors:
- do not redefine "j" three times, "ss" and "tsk_count_join" two
times, etc. Particuilarly i have no idea which "j" join you actually
would like to use. define the individual clauses youd like to use
once, then use that same instance. to SQLAlchemy, two identical
clauses are *not* interchangeable since it uses identity in many
cases with selectables.
- do *not* use orm.clear_mapper(). this method is gone in the next
version. use clear_mappers(), and redefine *all* mappers for each
test you are creating. mappers have lots of dependencies on each
other which are compiled when any of the mappers are first used.
there is no capability to surgically remove a mapper from that
compiled structure, you have to start over again.
- do *not* say, mapper.properties = {}. once a mapper is compiled,
the "properties" argument is not even looked at.
once all that is done, then ill have a better idea of what youre
actually trying to do (if it doesnt actually fix the problem).
On Jun 6, 2007, at 12:58 PM, [EMAIL PROTECTED] wrote:
> #!/usr/bin/env python
> from sqlalchemy import *
> import sys, datetime
>
> #init db
> #global_connect('mysql://test:[EMAIL PROTECTED]/test')
> #engine = create_engine('mysql://test:[EMAIL PROTECTED]/test')
>
> global_connect('sqlite:///tutorial.db')
> engine = create_engine('sqlite:///tutorial.db')
>
> project_t = Table('prj',
> Column('id', Integer,
> primary_key=True),
> Column('title', Unicode(100),
> nullable=False),
> mysql_engine='InnoDB')
>
>
> task_t = Table('task',
> Column('id', Integer,
> primary_key=True),
> Column('status_id', Integer,
> ForeignKey('task_status.id'), nullable=False),
> Column('title', Unicode(100),
> nullable=False),
> Column('task_type_id', Integer ,
> ForeignKey('task_type.id'), nullable=False),
> Column('prj_id', Integer ,
> ForeignKey('prj.id'), nullable=False),
> mysql_engine='InnoDB')
>
> task_status_t = Table('task_status',
> Column('id', Integer,
> primary_key=True),
> mysql_engine='InnoDB')
>
> task_type_t = Table('task_type',
> Column('id', Integer, primary_key=True),
> mysql_engine='InnoDB')
>
> message_t = Table('msg',
> Column('id', Integer, primary_key=True),
> Column('posted', DateTime, nullable=False,
> index=True, default=func.current_timestamp()),
> Column('type_id', Integer,
> ForeignKey('msg_type.id'), nullable=False, index=True),
> Column('from_uid', Integer, nullable=False,
> index=True),
> Column('to_uid', Integer, nullable=False,
> index=True),
> Column('task_id', Integer,
> ForeignKey('task.id'), nullable=True, index=True),
> Column('time_est_days', Integer, nullable=True),
> Column('subject', Unicode(60), nullable=True),
> Column('body', Unicode, nullable=True),
> Column('new', Boolean, nullable=False,
> default=True),
> Column('removed_by_sender', Boolean,
> nullable=False, default=False),
> Column('removed_by_recipient', Boolean,
> nullable=False, default=False),
> mysql_engine='InnoDB')
>
> message_type_t = Table('msg_type',
> Column('id', Integer,
> primary_key=True),
> Column('name', Unicode(20),
> nullable=False, unique=True),
> Column('display_name', Unicode(20),
> nullable=False, unique=True),
> mysql_engine='InnoDB')
>
> class Task(object):pass
>
> class Task_Type(object):pass
>
> class Message(object):pass
>
> class Message_Type(object):pass
>
> tsk_cnt_join = outerjoin(project_t, task_t,
> task_t.c.prj_id==project_t.c.id)
>
> ss = select([project_t.c.id.label('prj_id'),
> func.count(task_t.c.id).label('tasks_number')],
> from_obj=[tsk_cnt_join],
> group_by=[project_t.c.id]).alias('prj_tsk_cnt_s')
> j = join(project_t, ss, project_t.c.id == ss.c.prj_id)
>
> Task_Type.mapper = mapper(Task_Type, task_type_t)
>
>
> Task.mapper = mapper( Task, task_t,
> properties=dict(type=relation(Task_Type,
> lazy=False),
> ))
>
> Message_Type.mapper = mapper(Message_Type, message_type_t)
>
> Message.mapper = mapper(Message, message_t,
> properties=dict(type=relation(Message_Type,
> lazy=False, uselist=False),
> ))
>
> tsk_cnt_join = outerjoin(project_t, task_t,
> task_t.c.prj_id==project_t.c.id)
> ss = select([project_t.c.id.label('prj_id'),
> func.count(task_t.c.id).label('tasks_number')],
> from_obj=[tsk_cnt_join],
> group_by=[project_t.c.id]).alias('prj_tsk_cnt_s')
> j = join(project_t, ss, project_t.c.id == ss.c.prj_id)
>
> j = outerjoin( task_t, message_t, task_t.c.id==message_t.c.task_id)
> jj = select([ task_t.c.id.label('task_id'),
> func.count(message_t.c.id).label('props_cnt')],
> from_obj=[j], group_by=[task_t.c.id]).alias('prop_c_s')
> jjj = join(task_t, jj, task_t.c.id == jj.c.task_id)
>
> class cls(object):pass
>
> props =dict(type=relation(Task_Type, lazy=False))
> cls.mapper = mapper( cls, jjj, properties=props)
>
>
> default_metadata.engine.echo = True
> default_metadata.drop_all()
> default_metadata.create_all()
>
> session = create_session()
>
> engine.execute("INSERT INTO prj (title) values('project 1');")
> engine.execute("INSERT INTO task_status (id) values(1);")
> engine.execute("INSERT INTO task_type(id) values(1);")
> engine.execute("INSERT INTO task (title, task_type_id, status_id,
> prj_id) values('task 1',1,1,1);")
>
> #works
> cls.mapper.properties = {}
> for t in session.query(cls.mapper).limit(10).offset(0).list():
> print t.id, t.title, t.props_cnt
>
> #works
> for t in select([jjj], from_obj=[jjj],
> limit=10, offset=0).execute(bind_to=session.bind_to):
> print t.id, t.title, t.props_cnt
>
> orm.clear_mapper(cls.mapper)
> cls.mapper = mapper( cls, jjj, properties=props)
>
> #not works
> for t in session.query(cls.mapper).limit(10).offset(0).list():
> print t.id, t.title, t.props_cnt
--~--~---------~--~----~------------~-------~--~----~
You received this message because you are subscribed to the Google Groups
"sqlalchemy" group.
To post to this group, send email to sqlalchemy@googlegroups.com
To unsubscribe from this group, send email to [EMAIL PROTECTED]
For more options, visit this group at
http://groups.google.com/group/sqlalchemy?hl=en
-~----------~----~----~----~------~----~------~--~---
