root/galaxy-central/test/base/test_db_util.py

リビジョン 2, 8.3 KB (コミッタ: hatakeyama, 14 年 前)

import galaxy-central

行番号 
1import galaxy.model
2from galaxy.model.orm import *
3from galaxy.model.mapping import context as sa_session
4from base.twilltestcase import *
5import sys
6
7def delete_obj( obj ):
8    sa_session.delete( obj )
9    sa_session.flush()
10def delete_request_type_permissions( id ):
11    rtps = sa_session.query( galaxy.model.RequestTypePermissions ) \
12                     .filter( and_( galaxy.model.RequestTypePermissions.table.c.request_type_id==id ) ) \
13                     .order_by( desc( galaxy.model.RequestTypePermissions.table.c.create_time ) )
14    for rtp in rtps:
15        sa_session.delete( rtp )
16    sa_session.flush()
17def delete_user_roles( user ):
18    for ura in user.roles:
19        sa_session.delete( ura )
20    sa_session.flush()
21def flush( obj ):
22    sa_session.add( obj )
23    sa_session.flush()
24def get_all_histories_for_user( user ):
25    return sa_session.query( galaxy.model.History ) \
26                     .filter( and_( galaxy.model.History.table.c.user_id==user.id,
27                                    galaxy.model.History.table.c.deleted==False ) ) \
28                     .all()
29def get_dataset_permissions_by_dataset( dataset ):
30    return sa_session.query( galaxy.model.DatasetPermissions ) \
31                     .filter( galaxy.model.DatasetPermissions.table.c.dataset_id==dataset.id ) \
32                     .all()
33def get_dataset_permissions_by_role( role ):
34    return sa_session.query( galaxy.model.DatasetPermissions ) \
35                     .filter( galaxy.model.DatasetPermissions.table.c.role_id == role.id ) \
36                     .first()
37def get_default_history_permissions_by_history( history ):
38    return sa_session.query( galaxy.model.DefaultHistoryPermissions ) \
39                     .filter( galaxy.model.DefaultHistoryPermissions.table.c.history_id==history.id ) \
40                     .all()
41def get_default_history_permissions_by_role( role ):
42    return sa_session.query( galaxy.model.DefaultHistoryPermissions ) \
43                     .filter( galaxy.model.DefaultHistoryPermissions.table.c.role_id == role.id ) \
44                     .all()
45def get_default_user_permissions_by_role( role ):
46    return sa_session.query( galaxy.model.DefaultUserPermissions ) \
47                     .filter( galaxy.model.DefaultUserPermissions.table.c.role_id == role.id ) \
48                     .all()
49def get_default_user_permissions_by_user( user ):
50    return sa_session.query( galaxy.model.DefaultUserPermissions ) \
51                     .filter( galaxy.model.DefaultUserPermissions.table.c.user_id==user.id ) \
52                     .all()
53def get_form( name ):
54    fdc_list = sa_session.query( galaxy.model.FormDefinitionCurrent ) \
55                         .filter( galaxy.model.FormDefinitionCurrent.table.c.deleted == False ) \
56                         .order_by( galaxy.model.FormDefinitionCurrent.table.c.create_time.desc() )
57    for fdc in fdc_list:
58        sa_session.refresh( fdc )
59        sa_session.refresh( fdc.latest_form )
60        if fdc.latest_form.name == name:
61            return fdc.latest_form
62    return None
63def get_folder( parent_id, name, description ):
64    return sa_session.query( galaxy.model.LibraryFolder ) \
65                     .filter( and_( galaxy.model.LibraryFolder.table.c.parent_id==parent_id,
66                                    galaxy.model.LibraryFolder.table.c.name==name,
67                                    galaxy.model.LibraryFolder.table.c.description==description ) ) \
68                     .first()
69def get_group_by_name( name ):
70    return sa_session.query( galaxy.model.Group ).filter( galaxy.model.Group.table.c.name==name ).first()
71def get_group_role_associations_by_group( group ):
72    return sa_session.query( galaxy.model.GroupRoleAssociation ) \
73                     .filter( galaxy.model.GroupRoleAssociation.table.c.group_id == group.id ) \
74                     .all()
75def get_group_role_associations_by_role( role ):
76    return sa_session.query( galaxy.model.GroupRoleAssociation ) \
77                     .filter( galaxy.model.GroupRoleAssociation.table.c.role_id == role.id ) \
78                     .all()
79def get_latest_dataset():
80    return sa_session.query( galaxy.model.Dataset ) \
81                     .order_by( desc( galaxy.model.Dataset.table.c.create_time ) ) \
82                     .first()
83def get_latest_hda():
84    return sa_session.query( galaxy.model.HistoryDatasetAssociation ) \
85                     .order_by( desc( galaxy.model.HistoryDatasetAssociation.table.c.create_time ) ) \
86                     .first()
87def get_latest_history_for_user( user ):
88    return sa_session.query( galaxy.model.History ) \
89                     .filter( and_( galaxy.model.History.table.c.deleted==False,
90                                    galaxy.model.History.table.c.user_id==user.id ) ) \
91                     .order_by( desc( galaxy.model.History.table.c.create_time ) ) \
92                     .first()
93def get_latest_ldda_by_name( name ):
94    return sa_session.query( galaxy.model.LibraryDatasetDatasetAssociation ) \
95                     .filter( and_( galaxy.model.LibraryDatasetDatasetAssociation.table.c.name==name,
96                                    galaxy.model.LibraryDatasetDatasetAssociation.table.c.deleted == False ) ) \
97                     .order_by( desc( galaxy.model.LibraryDatasetDatasetAssociation.table.c.create_time ) ) \
98                     .first()
99def get_latest_lddas( limit ):
100    return sa_session.query( galaxy.model.LibraryDatasetDatasetAssociation ) \
101                     .order_by( desc( galaxy.model.LibraryDatasetDatasetAssociation.table.c.update_time ) ) \
102                     .limit( limit )
103def get_library( name, description, synopsis ):
104    return sa_session.query( galaxy.model.Library ) \
105                     .filter( and_( galaxy.model.Library.table.c.name==name,
106                                    galaxy.model.Library.table.c.description==description,
107                                    galaxy.model.Library.table.c.synopsis==synopsis,
108                                    galaxy.model.Library.table.c.deleted==False ) ) \
109                     .first()
110def get_private_role( user ):
111    for role in user.all_roles():
112        if role.name == user.email and role.description == 'Private Role for %s' % user.email:
113            return role
114    raise AssertionError( "Private role not found for user '%s'" % user.email )
115def get_request_by_name( name ):
116    return sa_session.query( galaxy.model.Request ) \
117                     .filter( and_( galaxy.model.Request.table.c.name==name,
118                                    galaxy.model.Request.table.c.deleted==False ) ) \
119                     .first()   
120def get_request_type_by_name( name ):
121    return sa_session.query( galaxy.model.RequestType ) \
122                     .filter( and_( galaxy.model.RequestType.table.c.name==name ) ) \
123                     .order_by( desc( galaxy.model.RequestType.table.c.create_time ) ) \
124                     .first()
125def get_role_by_name( name ):
126    return sa_session.query( galaxy.model.Role ).filter( galaxy.model.Role.table.c.name==name ).first()
127def get_user( email ):
128    return sa_session.query( galaxy.model.User ) \
129                     .filter( galaxy.model.User.table.c.email==email ) \
130                     .first()
131def get_user_address( user, short_desc ):
132    return sa_session.query( galaxy.model.UserAddress ) \
133                     .filter( and_( galaxy.model.UserAddress.table.c.user_id==user.id,
134                                    galaxy.model.UserAddress.table.c.desc==short_desc,
135                                    galaxy.model.UserAddress.table.c.deleted==False ) ) \
136                     .order_by( desc( galaxy.model.UserAddress.table.c.create_time ) ) \
137                     .first()   
138def get_user_group_associations_by_group( group ):
139    return sa_session.query( galaxy.model.UserGroupAssociation ) \
140                     .filter( galaxy.model.UserGroupAssociation.table.c.group_id == group.id ) \
141                     .all()
142def get_user_info_form_definition():
143    return galaxy.model.FormDefinition.types.USER_INFO
144def get_user_role_associations_by_role( role ):
145    return sa_session.query( galaxy.model.UserRoleAssociation ) \
146                     .filter( galaxy.model.UserRoleAssociation.table.c.role_id == role.id ) \
147                     .all()
148def mark_obj_deleted( obj ):
149    obj.deleted = True
150    sa_session.add( obj )
151    sa_session.flush()
152def refresh( obj ):
153    sa_session.refresh( obj )
Note: リポジトリブラウザについてのヘルプは TracBrowser を参照してください。