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

Code Duplication Length = 13-14 lines in 3 locations

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()

pyspider/database/sqlalchemy/resultdb.py 2 locations


        else:
            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):


        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))


		@@ 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()

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

binux / pyspider

Code Duplication Length = 13-14 lines in 3 locations

pyspider/database/sqlalchemy/taskdb.py 1 location

pyspider/database/sqlalchemy/resultdb.py 2 locations