source: use pygit2 instead of a subprocess to find the latest revision

cogs/meta.py

@@ -21,7 +21,9 @@
 
 import discord
 from discord.ext import commands
+import pygit2
 
+from bot import BASE_DIR
 import utils
 
 class Meta(commands.Cog):
@@ -73,7 +75,7 @@ async def source(self, context, *, command: str = None):
 		if module.startswith(self.__module__.split('.')[0]):  # XXX dunno if this branch works
 			# not a built-in command
 			location = os.path.relpath(inspect.getfile(src)).replace('\\', '/')
-			at = await self._current_revision()
+			at = self._current_revision()
 		elif module.startswith('discord'):
 			source_url = 'https://github.com/Rapptz/discord.py'
 			at = self._discord_revision()
@@ -91,12 +93,10 @@ async def source(self, context, *, command: str = None):
 		await context.send(final_url)
 
 	@staticmethod
-	@utils.asyncexecutor()
-	def _current_revision(*, default='master'):
-		try:
-			return os.popen('git rev-parse HEAD').read().strip()
-		except OSError:
-			return default
+	def _current_revision():
+		repo = pygit2.Repository(os.path.join(BASE_DIR, '.git'))
+		c = next(repo.walk(repo.head.target, pygit2.GIT_SORT_TOPOLOGICAL))
+		return c.hex[:6]
 
 	@classmethod
 	def _discord_revision(cls, *, default='rewrite'):

requirements.txt

@@ -2,3 +2,4 @@ asyncpg
 ben_cogs
 discord.py>=1.0.1,<2.0.0
 json5
+pygit2

utils.py

@@ -49,23 +49,6 @@ def __setitem__(self, key, value):
 
 		super().__setitem__(key, value)
 
-def asyncexecutor(*, timeout=None, loop=None, executor=None):
-	"""decorator that turns a synchronous function into an async one
-
-	Created by @Arqm#9302 (ID 325012556940836864). XXX Unknown license
-	"""
-	def decorator(func):
-		@functools.wraps(func)
-		def wrapper(*args, **kwargs):
-			partial = functools.partial(func, *args, **kwargs)
-			loop_ = loop or asyncio.get_event_loop()
-
-			coro = loop_.run_in_executor(executor, partial)
-			# don't need to check if timeout is None since wait_for will just "block" in that case anyway
-			return asyncio.wait_for(coro, timeout=timeout, loop=loop_)
-		return wrapper
-	return decorator
-
 class Guild(commands.Converter):
 	@staticmethod
 	async def convert(context, argument):