1 # GNU MediaGoblin -- federated, autonomous media hosting
2 # Copyright (C) 2011, 2012 MediaGoblin contributors. See AUTHORS.
4 # This program is free software: you can redistribute it and/or modify
5 # it under the terms of the GNU Affero General Public License as published by
6 # the Free Software Foundation, either version 3 of the License, or
7 # (at your option) any later version.
9 # This program is distributed in the hope that it will be useful,
10 # but WITHOUT ANY WARRANTY; without even the implied warranty of
11 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
12 # GNU Affero General Public License for more details.
14 # You should have received a copy of the GNU Affero General Public License
15 # along with this program. If not, see <http://www.gnu.org/licenses/>.
17 from mediagoblin
.db
.base
import Session
18 from mediagoblin
.db
.models
import MediaEntry
, Tag
, MediaTag
, Collection
21 ##########################
22 # Random utility functions
23 ##########################
26 def atomic_update(table
, query_dict
, update_values
):
27 table
.query
.filter_by(**query_dict
).update(update_values
,
28 synchronize_session
=False)
32 def check_media_slug_used(uploader_id
, slug
, ignore_m_id
):
33 query
= MediaEntry
.query
.filter_by(uploader
=uploader_id
, slug
=slug
)
34 if ignore_m_id
is not None:
35 query
= query
.filter(MediaEntry
.id != ignore_m_id
)
36 does_exist
= query
.first() is not None
40 def media_entries_for_tag_slug(dummy_db
, tag_slug
):
41 return MediaEntry
.query \
42 .join(MediaEntry
.tags_helper
) \
43 .join(MediaTag
.tag_helper
) \
45 (MediaEntry
.state
== u
'processed')
46 & (Tag
.slug
== tag_slug
))
49 def clean_orphan_tags(commit
=True):
50 """Search for unused MediaTags and delete them"""
51 q1
= Session
.query(Tag
).outerjoin(MediaTag
).filter(MediaTag
.id==None)
54 # The "let the db do all the work" version:
55 # q1 = Session.query(Tag.id).outerjoin(MediaTag).filter(MediaTag.id==None)
56 # q2 = Session.query(Tag).filter(Tag.id.in_(q1))
57 # q2.delete(synchronize_session = False)
62 def check_collection_slug_used(creator_id
, slug
, ignore_c_id
):
63 filt
= (Collection
.creator
== creator_id
) \
64 & (Collection
.slug
== slug
)
65 if ignore_c_id
is not None:
66 filt
= filt
& (Collection
.id != ignore_c_id
)
67 does_exist
= Session
.query(Collection
.id).filter(filt
).first() is not None
71 if __name__
== '__main__':
72 from mediagoblin
.db
.open import setup_connection_and_db_from_config
74 db
= setup_connection_and_db_from_config({'sql_engine':'sqlite:///mediagoblin.db'})