Code Duplication - binux/pyspider - Measure and Improve Code Quality continuously with Scrutinizer

Code Duplication Length = 13-14 lines in 3 locations

pyspider/database/sqlalchemy/resultdb.py 2 locations


            return self.engine.execute(self.table.insert()
                                       .values(**self._stringify(obj)))

    def select(self, project, fields=None, offset=0, limit=None):
        if project not in self.projects:
            self._list_project()
        if project not in self.projects:
            return
        self.table.name = self._tablename(project)

        columns = [getattr(self.table.c, f, f) for f in fields] if fields else self.table.c
        for task in self.engine.execute(self.table.select()
                                        .with_only_columns(columns=columns)
                                        .order_by(self.table.c.updatetime.desc())
                                        .offset(offset).limit(limit)
                                        .execution_options(autocommit=True)):
            yield self._parse(result2dict(columns, task))

    def count(self, project):
        if project not in self.projects:

        for count, in self.engine.execute(self.table.count()):
            return count

    def get(self, project, taskid, fields=None):
        if project not in self.projects:
            self._list_project()
        if project not in self.projects:
            return
        self.table.name = self._tablename(project)

        columns = [getattr(self.table.c, f, f) for f in fields] if fields else self.table.c
        for task in self.engine.execute(self.table.select()
                                        .with_only_columns(columns=columns)
                                        .where(self.table.c.taskid == taskid)
                                        .limit(1)):
            return self._parse(result2dict(columns, task))


pyspider/database/sqlalchemy/taskdb.py 1 location


                                            .where(self.table.c.status == status)):
                yield self._parse(result2dict(columns, task))

    def get_task(self, project, taskid, fields=None):
        if project not in self.projects:
            self._list_project()
        if project not in self.projects:
            return None

        self.table.name = self._tablename(project)
        columns = [getattr(self.table.c, f, f) for f in fields] if fields else self.table.c
        for each in self.engine.execute(self.table.select()
                                        .with_only_columns(columns)
                                        .limit(1)
                                        .where(self.table.c.taskid == taskid)):
            return self._parse(result2dict(columns, each))

    def status_count(self, project):
        result = dict()

		@@ 93-106 (lines=14) @@
90		return self.engine.execute(self.table.insert()
91		.values(**self._stringify(obj)))
92
93		def select(self, project, fields=None, offset=0, limit=None):
94		if project not in self.projects:
95		self._list_project()
96		if project not in self.projects:
97		return
98		self.table.name = self._tablename(project)
99
100		columns = [getattr(self.table.c, f, f) for f in fields] if fields else self.table.c
101		for task in self.engine.execute(self.table.select()
102		.with_only_columns(columns=columns)
103		.order_by(self.table.c.updatetime.desc())
104		.offset(offset).limit(limit)
105		.execution_options(autocommit=True)):
106		yield self._parse(result2dict(columns, task))
107
108		def count(self, project):
109		if project not in self.projects:
		@@ 118-130 (lines=13) @@
115		for count, in self.engine.execute(self.table.count()):
116		return count
117
118		def get(self, project, taskid, fields=None):
119		if project not in self.projects:
120		self._list_project()
121		if project not in self.projects:
122		return
123		self.table.name = self._tablename(project)
124
125		columns = [getattr(self.table.c, f, f) for f in fields] if fields else self.table.c
126		for task in self.engine.execute(self.table.select()
127		.with_only_columns(columns=columns)
128		.where(self.table.c.taskid == taskid)
129		.limit(1)):
130		return self._parse(result2dict(columns, task))
131

		@@ 105-117 (lines=13) @@
102		.where(self.table.c.status == status)):
103		yield self._parse(result2dict(columns, task))
104
105		def get_task(self, project, taskid, fields=None):
106		if project not in self.projects:
107		self._list_project()
108		if project not in self.projects:
109		return None
110
111		self.table.name = self._tablename(project)
112		columns = [getattr(self.table.c, f, f) for f in fields] if fields else self.table.c
113		for each in self.engine.execute(self.table.select()
114		.with_only_columns(columns)
115		.limit(1)
116		.where(self.table.c.taskid == taskid)):
117		return self._parse(result2dict(columns, each))
118
119		def status_count(self, project):
120		result = dict()

binux / pyspider

Code Duplication Length = 13-14 lines in 3 locations

pyspider/database/sqlalchemy/resultdb.py 2 locations

pyspider/database/sqlalchemy/taskdb.py 1 location