Commits

Frank Becker  committed f9ac3d3

refactoring, now you get data back

  • Participants
  • Parent commits 02d2813

Comments (0)

Files changed (8)

File couchdb/__init__.py

-#!/usr/bin/env python
-
-# -*- coding: utf-8 -*-
-
-"""
-Title
-"""
-__author__ = "Frank Becker <fb@alien8.de>"
-__version__ = "0.1"
-__date__ = "Date: 4/28/11"
-__copyright__ = "Copyright (c) 2011 Frank Becker"
-__license__ = "Python"
-
-  

File couchdb/_design/hosts/view/all/map.js

-/**
- * Created by .
- * User: a8
- * Date: 5/3/11
- * Time: 5:12 PM
- */
-function(doc) {
-    if (doc.doc_type == "Host")
-        emit(doc._id, doc);
-}

File couchdb/fill_couch.py

-#!/usr/bin/env python
-
-# -*- coding: utf-8 -*-
-
-"""
-Python source code - replace this with a description of the code and write
-the code below this text.
-"""
-__author__ = "Frank Becker <fb@alien8.de>"
-__version__ = "$Revision: 0.0 $"
-__date__ = "$Date: YDATE $"
-__copyright__ = "Copyright (c) 2010 Frank Becker"
-__license__ = "Python"
-
-from datetime import datetime
-import couchdbkit
-
-class Host(couchdbkit.Document):
-    """CouchDB doc for Host"""
-    name = couchdbkit.StringProperty()
-    date_updated = couchdbkit.DateTimeProperty()
-    date_created = couchdbkit.DateTimeProperty()
-
-
-def db_connect():
-    """connect to db"""
-
-    try:
-        server = couchdbkit.Server(uri="http://alzey.osrc.amd.com:5984")
-    except:
-        return False
-
-    db = server.get_or_create_db("myhost_hosts")
-
-    return db
-
-def db_register(doc_class, db):
-    """Register Document class to DB and return it
-
-    FIXME (a8): This doesn't work, I got no fucking clue why
-    """
-
-    try:
-        setattr(doc_class, 'set_db', db)
-    except:
-        return False
-
-    return doc_class
-
-def save_or_create_host(cls, name=None, **kwargs):
-    """
-
-    Args:
-        cls registered Host() class
-        name=None, **kwargs
-
-    Returns:
-
-        host obj
-
-    """
-    if not name:
-        return False
-
-    host = cls(name=name, date_created=datetime.utcnow())
-    for key, value in kwargs.items():
-        setattr(host, key, value)
-    try:
-        host.save()
-    except:
-        return False
-
-    return host
-
-def send_view(path):
-    """
-    send design document to db
-    """
-    couchdbkit.designer.push(path, db_connect())
-
-def get_all_hosts():
-    """
-
-    Args:
-
-
-    Returns:
-
-    """
-    return Host.view("hosts/all")
-
-def get_host_by_name(host):
-    """
-
-    Args:
-
-
-    Returns:
-
-    """
-    return Host.view("hosts/one", key=host)
-
-
-def main():
-    """ main
-    """
-    db_connect()
-
-if __name__ == '__main__':
-    main()
-
-# vim: tabstop=4 expandtab shiftwidth=4 softtabstop=4 fileencoding=utf-8 :
-

File couchdb/tests.py

-#!/usr/bin/env python
-
-# -*- coding: utf-8 -*-
-import couchdbkit
-
-__author__ = 'a8'
-
-import unittest
-import fill_couch
-
-class CouchDBConnect(unittest.TestCase):
-
-    def test_db_connect(self):
-        self.assertNotEqual(fill_couch.db_connect(), False)
-
-class CouchDBRegister(unittest.TestCase):
-
-    def setUp(self):
-        """
-        init db
-        """
-        self.db = fill_couch.db_connect()
-
-    def disabled_db_register(self):
-        self.assertTrue(issubclass(
-            fill_couch.db_register(fill_couch.Host, self.db),
-            couchdbkit.Document))
-
-    def tearDown(self):
-        del(self.db)
-
-
-class SaveHost(unittest.TestCase):
-    """"""
-
-    def setUp(self):
-        """
-        init db
-        """
-        self.db = fill_couch.db_connect()
-        fill_couch.Host.set_db(self.db)
-
-    def test_simple_save(self):
-        """"test to save a simple host"""
-        self.host1 = fill_couch.save_or_create_host(
-            fill_couch.Host,
-            name="eris")
-        self.assertTrue(isinstance(
-            self.host1,
-            fill_couch.Host))
-        self.host1.delete()
-
-    def test_complex_host_save(self):
-        """"test to save a simple host"""
-        self.host2 = fill_couch.save_or_create_host(
-            fill_couch.Host,
-            name="fnord",
-            cpu_family="fam10",
-            )
-        self.assertTrue(isinstance(
-            self.host2,
-            fill_couch.Host))
-        self.assertEqual(self.host2.name, "fnord")
-        self.assertEqual(self.host2.cpu_family, "fam10")
-        self.host2.delete()
-
-    def tearDown(self):
-        del(self.db)
-
-if __name__ == '__main__':
-    unittest.main()

File database/__init__.py

+#!/usr/bin/env python
+
+# -*- coding: utf-8 -*-
+
+"""
+Title
+"""
+__author__ = "Frank Becker <fb@alien8.de>"
+__version__ = "0.1"
+__date__ = "Date: 4/28/11"
+__copyright__ = "Copyright (c) 2011 Frank Becker"
+__license__ = "Python"
+
+  

File database/_design/hosts/views/all/map.js

+/**
+ * Created by .
+ * User: a8
+ * Date: 5/3/11
+ * Time: 5:12 PM
+ */
+function(doc) {
+    if (doc.doc_type == "Host")
+        emit(doc._id, doc);
+}

File database/couchdb.py

+#!/usr/bin/env python
+
+# -*- coding: utf-8 -*-
+
+"""
+Tools to deal with data from external data bases
+"""
+__author__ = "Frank Becker <fb@alien8.de>"
+__version__ = "$Revision: 0.0 $"
+__date__ = "$Date: YDATE $"
+__copyright__ = "Copyright (c) 2010 Frank Becker"
+__license__ = "Python"
+
+from datetime import datetime
+import couchdbkit
+
+class Host(couchdbkit.Document):
+    """CouchDB doc for Host"""
+
+    def save(self):
+        self.date_updated = datetime.utcnow()
+        if not self.date_created:
+            self.date_created = datetime.utcnow()
+        super(Host, self).save()
+
+    name = couchdbkit.StringProperty()
+    date_updated = couchdbkit.DateTimeProperty()
+    date_created = couchdbkit.DateTimeProperty()
+
+
+def db_connect():
+    """connect to db"""
+
+    try:
+        server = couchdbkit.Server(uri="http://alzey.osrc.amd.com:5984")
+    except:
+        return False
+
+    db = server.get_or_create_db("myhost_hosts")
+
+    return db
+
+def db_register(doc_class, db):
+    """Register Document class to DB and return it
+
+    FIXME (a8): This doesn't work, take care in what namespace Host() is
+    """
+
+    try:
+        setattr(doc_class, 'set_db', db)
+    except:
+        return False
+
+    return doc_class
+
+def save_or_create_host(cls, name=None, **kwargs):
+    """
+
+    Args:
+        cls registered Host() class
+        name=None, **kwargs
+
+    Returns:
+
+        host obj
+
+    """
+    if not name:
+        return False
+
+    host = cls(name=name, date_created=datetime.utcnow())
+    for key, value in kwargs.items():
+        setattr(host, key, value)
+    try:
+        host.save()
+    except:
+        return False
+
+    return host
+
+def send_view(path):
+    """
+    send design document to db
+    """
+    couchdbkit.designer.push(path, db_connect())
+
+def get_all_hosts():
+    """
+
+    Args:
+
+
+    Returns:
+
+    """
+    return Host.view("hosts/all")
+
+def get_hosts_by_filter(**kwargs):
+    """
+
+    Returns a list of filtered hosts. Make sure Host() is in the namespace and
+    it has a db set. The list is sorted by the last update of the host in the
+    db.
+
+    Example:
+    * all hosts w/ the name eris
+        get_hosts_by_filter(name='eris')
+    * the last updated hosts w/ the name eris
+        get_hosts_by_filter(name='eris')[0]
+
+    Args:
+        filter are key=value requirements
+
+
+    Returns:
+        list of hosts sorted by Host.date_updated
+
+    """
+
+    if kwargs:
+        hosts = Host.view("hosts/all").all()
+
+        for key, value in kwargs.items():
+            hosts = filter(
+                lambda h: hasattr(h, key) and getattr(h, key) == value, hosts
+            )
+    else:
+        hosts = []
+
+    return sorted(hosts, key=lambda host: host.date_updated)
+
+
+def main():
+    """ main
+    """
+    db_connect()
+
+if __name__ == '__main__':
+    main()
+
+# vim: tabstop=4 expandtab shiftwidth=4 softtabstop=4 fileencoding=utf-8 :
+

File database/tests.py

+#!/usr/bin/env python
+
+# -*- coding: utf-8 -*-
+import time
+import uuid
+import couchdbkit
+
+__author__ = 'a8'
+
+import unittest
+from couchdb import Host, get_all_hosts, get_hosts_by_filter, db_connect, \
+    db_register, save_or_create_host
+
+class CouchDBConnect(unittest.TestCase):
+
+    def test_db_connect(self):
+        self.assertNotEqual(db_connect(), False)
+
+class CouchDBRegister(unittest.TestCase):
+
+    def setUp(self):
+        """
+        init db
+        """
+        self.db = db_connect()
+
+    def disabled_db_register(self):
+        self.assertTrue(issubclass(
+            db_register(Host, self.db),
+            couchdbkit.Document))
+
+    def tearDown(self):
+        del(self.db)
+
+
+class SaveHost(unittest.TestCase):
+    """"""
+
+    def setUp(self):
+        """
+        init db
+        """
+        self.db = db_connect()
+        Host.set_db(self.db)
+
+    def test_simple_save(self):
+        """"test to save a simple host"""
+        self.host1 = save_or_create_host(
+            Host,
+            name="eris")
+        self.assertTrue(isinstance(
+            self.host1,
+            Host))
+        self.host1.delete()
+
+    def test_complex_host_save(self):
+        """"test to save a simple host"""
+        self.host2 = save_or_create_host(
+            Host,
+            name="fnord",
+            cpu_family="fam10",
+            )
+        self.assertTrue(isinstance(
+            self.host2,
+            Host))
+        self.assertEqual(self.host2.name, "fnord")
+        self.assertEqual(self.host2.cpu_family, "fam10")
+        self.host2.delete()
+
+    def tearDown(self):
+        del(self.db)
+
+
+class GetHosts(unittest.TestCase):
+    """"""
+
+    def setUp(self):
+        """
+        init db
+        """
+        self.db = db_connect()
+        Host.set_db(self.db)
+        self.host1 = save_or_create_host(
+            Host,
+            name="eris")
+        self.host2 = save_or_create_host(
+            Host,
+            name="fridolin")
+        self.uuid3 = str(uuid.uuid4())
+        self.host3 = save_or_create_host(
+            Host,
+            name="fridolin",
+            uuid=self.uuid3,
+        )
+
+    def test_get_hosts(self):
+        """"test to save a simple host"""
+        self.assertTrue(len(Host.view("hosts/all").all()) >= 2)
+
+    def test_get_hosts_by_name(self):
+        """"test to save a simple host"""
+        self.assertTrue(
+            len(get_hosts_by_filter(name="fridolin")) >= 2
+        )
+
+    def test_get_hosts_by_name(self):
+        """"test to save a simple host"""
+        self.assertTrue(
+            len(get_hosts_by_filter(name="fridolin", uuid=self.uuid3)) == 1
+        )
+
+    def test_get_hosts_by_name_sorted(self):
+        """"test to save a simple host"""
+        time.sleep(2)
+        # set update time
+        self.host3.save()
+        host = get_hosts_by_filter(name="fridolin")[-1]
+        self.assertTrue(getattr(host, 'uuid') == self.uuid3)
+
+    def tearDown(self):
+        self.host1.delete()
+        self.host2.delete()
+        self.host3.delete()
+        del self.db
+
+if __name__ == '__main__':
+    unittest.main()