daklib/projectb.py

   1 #!/usr/bin/python
   2
   3 """
   4 Class providing access to a projectb database
   5
   6 This class provides convenience functions for common queries to a
   7 projectb database using psycopg2.
   8
   9 Copyright (C) 2009  Mike O'Connor <stew@vireo.org>
  10 """
  11
  12 # This program is free software; you can redistribute it and/or modify
  13 # it under the terms of the GNU General Public License as published by
  14 # the Free Software Foundation; either version 2 of the License, or
  15 # (at your option) any later version.
  16
  17 # This program is distributed in the hope that it will be useful,
  18 # but WITHOUT ANY WARRANTY; without even the implied warranty of
  19 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  20 # GNU General Public License for more details.
  21
  22 # You should have received a copy of the GNU General Public License
  23 # along with this program; if not, write to the Free Software
  24 # Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
  25
  26 ################################################################################
  27
  28 import psycopg2
  29
  30 ################################################################################
  31
  32 class Projectb(object):
  33     """
  34     Object providing methods for accessing the projectb database
  35     """
  36     def __init__(self,Cnf):
  37         connect_str = "dbname=%s"% (Cnf["DB::Name"])
  38         if Cnf["DB::Host"] != '': connect_str += " host=%s" % (Cnf["DB::Host"])
  39         if Cnf["DB::Port"] != '-1': connect_str += " port=%d" % (int(Cnf["DB::Port"]))
  40
  41         self.dbh = psycopg2.connect(connect_str)
  42         self.suite_id_cache = {}
  43         self.architecture_id_cache = {}
  44         self.section_id_cache = {}
  45
  46     def get_suite_id(self, suite_name):
  47         """
  48         return the id for the given suite_name
  49
  50         @param suite_name: name of a suite such as "unsatble" or "testing"
  51
  52         @rtype: int
  53         @return: id of given suite or None if suite_name not matched
  54
  55         >>> Cnf = {'DB::Name' : "projectb","DB::Host":"","DB::Port":'-1' }
  56         >>> pb = Projectb( Cnf )
  57         >>> pb.get_suite_id("unstable")
  58         5
  59         >>> pb.get_suite_id("n'existe pas")
  60         """
  61         if not self.suite_id_cache.has_key(suite_name):
  62             c = self.dbh.cursor()
  63             c.execute("SELECT id FROM suite WHERE suite_name=%(suite_name)s",
  64                       {'suite_name':suite_name})
  65             r = c.fetchone()
  66             if r:
  67                 self.suite_id_cache[suite_name] = r[0]
  68             else:
  69                 self.suite_id_cache[suite_name] = None
  70
  71         return self.suite_id_cache[suite_name]
  72
  73     def get_architecture_id(self, architecture_name):
  74         """
  75         return the id for the given architecture_name
  76
  77         @param architecture_name: name of a architecture such as "i386" or "source"
  78
  79         @rtype: int
  80         @return: id of given architecture or None if architecture_name not matched
  81
  82         >>> Cnf = {'DB::Name' : "projectb","DB::Host":"","DB::Port":'-1' }
  83         >>> pb = Projectb( Cnf )
  84         >>> pb.get_architecture_id("i386")
  85         7
  86         >>> pb.get_architecture_id("n'existe pas")
  87         """
  88         if not self.architecture_id_cache.has_key(architecture_name):
  89             c = self.dbh.cursor()
  90             c.execute("SELECT id FROM architecture WHERE arch_string=%(architecture_name)s",
  91                       {'architecture_name':architecture_name})
  92             r = c.fetchone()
  93             if r:
  94                 self.architecture_id_cache[architecture_name] = r[0]
  95             else:
  96                 self.architecture_id_cache[architecture_name] = None
  97
  98         return self.architecture_id_cache[architecture_name]
  99
 100     def get_section_id(self, section_name):
 101         """
 102         return the id for the given section_name
 103
 104         @param section_name: name of a section such as "x11" or "non-free/libs"
 105
 106         @rtype: int
 107         @return: id of given section or None if section_name not matched
 108
 109         >>> Cnf = {'DB::Name' : "projectb","DB::Host":"","DB::Port":'-1' }
 110         >>> pb = Projectb( Cnf )
 111         >>> pb.get_section_id("non-free/libs")
 112         285
 113         >>> pb.get_section_id("n'existe pas")
 114         """
 115         if not self.section_id_cache.has_key(section_name):
 116             c = self.dbh.cursor()
 117             c.execute("SELECT id FROM section WHERE section=%(section_name)s",
 118                       {'section_name':section_name})
 119             r = c.fetchone()
 120             if r:
 121                 self.section_id_cache[section_name] = r[0]
 122             else:
 123                 self.section_id_cache[section_name] = None
 124
 125         return self.section_id_cache[section_name]
 126
 127 if __name__ == "__main__":
 128     import doctest
 129     doctest.testmod()