Changeset - 8c99d5d8eb0e
Parent rev.
Child rev.
[Not reviewed]
default
0 1 0
Mads Kiilerich - 6 years ago 2020-01-02 00:57:38
mads@kiilerich.com
Grafted from: 7c11d8ff3bf7
cleanup: fix db.py "import *"

pyflakes rightfully complains about wildcard imports. But we like to keep the
short names, so let's just be explicit about which names we import.
1 file changed with 10 insertions and 10 deletions:
kallithea/model/db.py
10
10
0 comments (0 inline, 0 general)
kallithea/model/db.py
➞
Show inline comments
 
@@ -29,25 +29,25 @@ import base64
 
import collections
 
import datetime
 
import functools
 
import hashlib
 
import logging
 
import os
 
import time
 
import traceback
 

			




	
	
	
		
 
import ipaddr

			




	
	
	
		
 
import sqlalchemy

			




	
	
	
		
 
from beaker.cache import cache_region, region_invalidate

			




	
	
	
		
 
from sqlalchemy import *

			




	
	
	
		
 
from sqlalchemy import Boolean, Column, DateTime, Float, ForeignKey, Index, Integer, LargeBinary, String, Unicode, UnicodeText, UniqueConstraint

			




	
	
	
		
 
from sqlalchemy.ext.hybrid import hybrid_property

			




	
	
	
		
 
from sqlalchemy.orm import class_mapper, joinedload, relationship, validates

			




	
	
	
		
 
from tg.i18n import lazy_ugettext as _

			




	
	
	
		
 
from webob.exc import HTTPNotFound

			




	
	
	
		
 

			




	
	
	
		
 
import kallithea

			




	
	
	
		
 
from kallithea.lib.caching_query import FromCache

			




	
	
	
		
 
from kallithea.lib.compat import json

			




	
	
	
		
 
from kallithea.lib.exceptions import DefaultUserException

			




	
	
	
		
 
from kallithea.lib.utils2 import (

			




	
	
	
		
 
    Optional, aslist, get_changeset_safe, get_clone_url, remove_prefix, safe_bytes, safe_int, safe_str, safe_unicode, str2bool, urlreadable)

			




	
	
	
		
 
from kallithea.lib.vcs import get_backend

			




	
	
		
 
@@ -558,25 +558,25 @@ class User(Base, BaseDbModel):

			




	
	
	
		
 
        For anything else, try to look up by the user name.

			




	
	
	
		
 

			




	
	
	
		
 
        This assumes no normal username can have '@' symbol.

			




	
	
	
		
 
        """

			




	
	
	
		
 
        if '@' in username_or_email:

			




	
	
	
		
 
            return User.get_by_email(username_or_email, cache=cache)

			




	
	
	
		
 
        else:

			




	
	
	
		
 
            return User.get_by_username(username_or_email, case_insensitive=case_insensitive, cache=cache)

			




	
	
	
		
 

			




	
	
	
		
 
    @classmethod

			




	
	
	
		
 
    def get_by_username(cls, username, case_insensitive=False, cache=False):

			




	
	
	
		
 
        if case_insensitive:

			




	
	
	
		
 
            q = cls.query().filter(func.lower(cls.username) == func.lower(username))

			




	
	
	
		
 
            q = cls.query().filter(sqlalchemy.func.lower(cls.username) == sqlalchemy.func.lower(username))

			




	
	
	
		
 
        else:

			




	
	
	
		
 
            q = cls.query().filter(cls.username == username)

			




	
	
	
		
 

			




	
	
	
		
 
        if cache:

			




	
	
	
		
 
            q = q.options(FromCache(

			




	
	
	
		
 
                            "sql_cache_short",

			




	
	
	
		
 
                            "get_user_%s" % _hash_key(username)

			




	
	
	
		
 
                          )

			




	
	
	
		
 
            )

			




	
	
	
		
 
        return q.scalar()

			




	
	
	
		
 

			




	
	
	
		
 
    @classmethod

			




	
	
		
 
@@ -593,35 +593,35 @@ class User(Base, BaseDbModel):

			




	
	
	
		
 

			




	
	
	
		
 
        if fallback and not res:

			




	
	
	
		
 
            # fallback to additional keys

			




	
	
	
		
 
            _res = UserApiKeys.query().filter_by(api_key=api_key, is_expired=False).first()

			




	
	
	
		
 
            if _res:

			




	
	
	
		
 
                res = _res.user

			




	
	
	
		
 
        if res is None or not res.active or res.is_default_user:

			




	
	
	
		
 
            return None

			




	
	
	
		
 
        return res

			




	
	
	
		
 

			




	
	
	
		
 
    @classmethod

			




	
	
	
		
 
    def get_by_email(cls, email, cache=False):

			




	
	
	
		
 
        q = cls.query().filter(func.lower(cls.email) == func.lower(email))

			




	
	
	
		
 
        q = cls.query().filter(sqlalchemy.func.lower(cls.email) == sqlalchemy.func.lower(email))

			




	
	
	
		
 

			




	
	
	
		
 
        if cache:

			




	
	
	
		
 
            q = q.options(FromCache("sql_cache_short",

			




	
	
	
		
 
                                    "get_email_key_%s" % email))

			




	
	
	
		
 

			




	
	
	
		
 
        ret = q.scalar()

			




	
	
	
		
 
        if ret is None:

			




	
	
	
		
 
            q = UserEmailMap.query()

			




	
	
	
		
 
            # try fetching in alternate email map

			




	
	
	
		
 
            q = q.filter(func.lower(UserEmailMap.email) == func.lower(email))

			




	
	
	
		
 
            q = q.filter(sqlalchemy.func.lower(UserEmailMap.email) == sqlalchemy.func.lower(email))

			




	
	
	
		
 
            q = q.options(joinedload(UserEmailMap.user))

			




	
	
	
		
 
            if cache:

			




	
	
	
		
 
                q = q.options(FromCache("sql_cache_short",

			




	
	
	
		
 
                                        "get_email_map_key_%s" % email))

			




	
	
	
		
 
            ret = getattr(q.scalar(), 'user', None)

			




	
	
	
		
 

			




	
	
	
		
 
        return ret

			




	
	
	
		
 

			




	
	
	
		
 
    @classmethod

			




	
	
	
		
 
    def get_from_cs_author(cls, author):

			




	
	
	
		
 
        """

			




	
	
	
		
 
        Tries to get User objects out of commit author string

			




	
	
		
 
@@ -848,25 +848,25 @@ class UserGroup(Base, BaseDbModel):

			




	
	
	
		
 
        return "<%s %s: %r')>" % (self.__class__.__name__,

			




	
	
	
		
 
                                  self.users_group_id,

			




	
	
	
		
 
                                  self.users_group_name)

			




	
	
	
		
 

			




	
	
	
		
 
    @classmethod

			




	
	
	
		
 
    def guess_instance(cls, value):

			




	
	
	
		
 
        return super(UserGroup, cls).guess_instance(value, UserGroup.get_by_group_name)

			




	
	
	
		
 

			




	
	
	
		
 
    @classmethod

			




	
	
	
		
 
    def get_by_group_name(cls, group_name, cache=False,

			




	
	
	
		
 
                          case_insensitive=False):

			




	
	
	
		
 
        if case_insensitive:

			




	
	
	
		
 
            q = cls.query().filter(func.lower(cls.users_group_name) == func.lower(group_name))

			




	
	
	
		
 
            q = cls.query().filter(sqlalchemy.func.lower(cls.users_group_name) == sqlalchemy.func.lower(group_name))

			




	
	
	
		
 
        else:

			




	
	
	
		
 
            q = cls.query().filter(cls.users_group_name == group_name)

			




	
	
	
		
 
        if cache:

			




	
	
	
		
 
            q = q.options(FromCache(

			




	
	
	
		
 
                            "sql_cache_short",

			




	
	
	
		
 
                            "get_group_%s" % _hash_key(group_name)

			




	
	
	
		
 
                          )

			




	
	
	
		
 
            )

			




	
	
	
		
 
        return q.scalar()

			




	
	
	
		
 

			




	
	
	
		
 
    @classmethod

			




	
	
	
		
 
    def get(cls, user_group_id, cache=False):

			




	
	
		
 
@@ -1045,25 +1045,25 @@ class Repository(Base, BaseDbModel):

			




	
	
	
		
 
        except Exception:

			




	
	
	
		
 
            log.error(traceback.format_exc())

			




	
	
	
		
 

			




	
	
	
		
 
    @classmethod

			




	
	
	
		
 
    def query(cls, sorted=False):

			




	
	
	
		
 
        """Add Repository-specific helpers for common query constructs.

			




	
	
	
		
 

			




	
	
	
		
 
        sorted: if True, apply the default ordering (name, case insensitive).

			




	
	
	
		
 
        """

			




	
	
	
		
 
        q = super(Repository, cls).query()

			




	
	
	
		
 

			




	
	
	
		
 
        if sorted:

			




	
	
	
		
 
            q = q.order_by(func.lower(Repository.repo_name))

			




	
	
	
		
 
            q = q.order_by(sqlalchemy.func.lower(Repository.repo_name))

			




	
	
	
		
 

			




	
	
	
		
 
        return q

			




	
	
	
		
 

			




	
	
	
		
 
    @classmethod

			




	
	
	
		
 
    def url_sep(cls):

			




	
	
	
		
 
        return URL_SEP

			




	
	
	
		
 

			




	
	
	
		
 
    @classmethod

			




	
	
	
		
 
    def normalize_repo_name(cls, repo_name):

			




	
	
	
		
 
        """

			




	
	
	
		
 
        Normalizes os specific repo_name to the format internally stored inside

			




	
	
	
		
 
        database using URL_SEP

			




	
	
		
 
@@ -1073,25 +1073,25 @@ class Repository(Base, BaseDbModel):

			




	
	
	
		
 
        """

			




	
	
	
		
 
        return cls.url_sep().join(repo_name.split(os.sep))

			




	
	
	
		
 

			




	
	
	
		
 
    @classmethod

			




	
	
	
		
 
    def guess_instance(cls, value):

			




	
	
	
		
 
        return super(Repository, cls).guess_instance(value, Repository.get_by_repo_name)

			




	
	
	
		
 

			




	
	
	
		
 
    @classmethod

			




	
	
	
		
 
    def get_by_repo_name(cls, repo_name, case_insensitive=False):

			




	
	
	
		
 
        """Get the repo, defaulting to database case sensitivity.

			




	
	
	
		
 
        case_insensitive will be slower and should only be specified if necessary."""

			




	
	
	
		
 
        if case_insensitive:

			




	
	
	
		
 
            q = Session().query(cls).filter(func.lower(cls.repo_name) == func.lower(repo_name))

			




	
	
	
		
 
            q = Session().query(cls).filter(sqlalchemy.func.lower(cls.repo_name) == sqlalchemy.func.lower(repo_name))

			




	
	
	
		
 
        else:

			




	
	
	
		
 
            q = Session().query(cls).filter(cls.repo_name == repo_name)

			




	
	
	
		
 
        q = q.options(joinedload(Repository.fork)) \

			




	
	
	
		
 
                .options(joinedload(Repository.owner)) \

			




	
	
	
		
 
                .options(joinedload(Repository.group))

			




	
	
	
		
 
        return q.scalar()

			




	
	
	
		
 

			




	
	
	
		
 
    @classmethod

			




	
	
	
		
 
    def get_by_full_path(cls, repo_full_path):

			




	
	
	
		
 
        base_full_path = os.path.realpath(cls.base_path())

			




	
	
	
		
 
        repo_full_path = os.path.realpath(repo_full_path)

			




	
	
	
		
 
        assert repo_full_path.startswith(base_full_path + os.path.sep)

			




	
	
		
 
@@ -1466,25 +1466,25 @@ class RepoGroup(Base, BaseDbModel):

			




	
	
	
		
 
    parent_group = relationship('RepoGroup', remote_side=group_id)

			




	
	
	
		
 
    owner = relationship('User')

			




	
	
	
		
 

			




	
	
	
		
 
    @classmethod

			




	
	
	
		
 
    def query(cls, sorted=False):

			




	
	
	
		
 
        """Add RepoGroup-specific helpers for common query constructs.

			




	
	
	
		
 

			




	
	
	
		
 
        sorted: if True, apply the default ordering (name, case insensitive).

			




	
	
	
		
 
        """

			




	
	
	
		
 
        q = super(RepoGroup, cls).query()

			




	
	
	
		
 

			




	
	
	
		
 
        if sorted:

			




	
	
	
		
 
            q = q.order_by(func.lower(RepoGroup.group_name))

			




	
	
	
		
 
            q = q.order_by(sqlalchemy.func.lower(RepoGroup.group_name))

			




	
	
	
		
 

			




	
	
	
		
 
        return q

			




	
	
	
		
 

			




	
	
	
		
 
    def __init__(self, group_name='', parent_group=None):

			




	
	
	
		
 
        self.group_name = group_name

			




	
	
	
		
 
        self.parent_group = parent_group

			




	
	
	
		
 

			




	
	
	
		
 
    def __repr__(self):

			




	
	
	
		
 
        return "<%s %s: %s>" % (self.__class__.__name__,

			




	
	
	
		
 
                                self.group_id, self.group_name)

			




	
	
	
		
 

			




	
	
	
		
 
    @classmethod

			




	
	
		
 
@@ -1505,25 +1505,25 @@ class RepoGroup(Base, BaseDbModel):

			




	
	
	
		
 
    def url_sep(cls):

			




	
	
	
		
 
        return URL_SEP

			




	
	
	
		
 

			




	
	
	
		
 
    @classmethod

			




	
	
	
		
 
    def guess_instance(cls, value):

			




	
	
	
		
 
        return super(RepoGroup, cls).guess_instance(value, RepoGroup.get_by_group_name)

			




	
	
	
		
 

			




	
	
	
		
 
    @classmethod

			




	
	
	
		
 
    def get_by_group_name(cls, group_name, cache=False, case_insensitive=False):

			




	
	
	
		
 
        group_name = group_name.rstrip('/')

			




	
	
	
		
 
        if case_insensitive:

			




	
	
	
		
 
            gr = cls.query() \

			




	
	
	
		
 
                .filter(func.lower(cls.group_name) == func.lower(group_name))

			




	
	
	
		
 
                .filter(sqlalchemy.func.lower(cls.group_name) == sqlalchemy.func.lower(group_name))

			




	
	
	
		
 
        else:

			




	
	
	
		
 
            gr = cls.query() \

			




	
	
	
		
 
                .filter(cls.group_name == group_name)

			




	
	
	
		
 
        if cache:

			




	
	
	
		
 
            gr = gr.options(FromCache(

			




	
	
	
		
 
                            "sql_cache_short",

			




	
	
	
		
 
                            "get_group_%s" % _hash_key(group_name)

			




	
	
	
		
 
                            )

			




	
	
	
		
 
            )

			




	
	
	
		
 
        return gr.scalar()

			




	
	
	
		
 

			




	
	
	
		
 
    @property

			




	
	
		
 
@@ -2001,25 +2001,25 @@ class UserFollowing(Base, BaseDbModel):

			




	
	
	
		
 
        _table_args_default_dict,

			




	
	
	
		
 
    )

			




	
	
	
		
 

			




	
	
	
		
 
    user_following_id = Column(Integer(), primary_key=True)

			




	
	
	
		
 
    user_id = Column(Integer(), ForeignKey('users.user_id'), nullable=False)

			




	
	
	
		
 
    follows_repository_id = Column(Integer(), ForeignKey('repositories.repo_id'), nullable=True)

			




	
	
	
		
 
    follows_user_id = Column(Integer(), ForeignKey('users.user_id'), nullable=True)

			




	
	
	
		
 
    follows_from = Column(DateTime(timezone=False), nullable=False, default=datetime.datetime.now)

			




	
	
	
		
 

			




	
	
	
		
 
    user = relationship('User', primaryjoin='User.user_id==UserFollowing.user_id')

			




	
	
	
		
 

			




	
	
	
		
 
    follows_user = relationship('User', primaryjoin='User.user_id==UserFollowing.follows_user_id')

			




	
	
	
		
 
    follows_repository = relationship('Repository', order_by=lambda: func.lower(Repository.repo_name))

			




	
	
	
		
 
    follows_repository = relationship('Repository', order_by=lambda: sqlalchemy.func.lower(Repository.repo_name))

			




	
	
	
		
 

			




	
	
	
		
 
    @classmethod

			




	
	
	
		
 
    def get_repo_followers(cls, repo_id):

			




	
	
	
		
 
        return cls.query().filter(cls.follows_repository_id == repo_id)

			




	
	
	
		
 

			




	
	
	
		
 

			




	
	
	
		
 
class CacheInvalidation(Base, BaseDbModel):

			




	
	
	
		
 
    __tablename__ = 'cache_invalidation'

			




	
	
	
		
 
    __table_args__ = (

			




	
	
	
		
 
        Index('key_idx', 'cache_key'),

			




	
	
	
		
 
        _table_args_default_dict,

			




	
	
	
		
 
    )

			




        

    



    


    



    



      

      





    
    0 comments (0 inline, 0 general)
    





    


  

  







  


    




    








    
        
    
    
        This site is powered by
            Kallithea 0.7.0,
        which is
        © 2010–2025 by various authors & licensed under GPLv3.