Implement self-hosted Git LFS server on top Flask

git-lfs-self/requirements.txt

@@ -0,0 +1 @@
+flask

git-lfs-self/src/main/app.py

@@ -0,0 +1,158 @@
+import argparse
+import json
+import uuid
+from pathlib import Path
+
+import flask
+from flask import request
+from werkzeug import wsgi
+
+from core import *
+
+__all__ = (
+    'SimpleLargeFileStorage',
+    'Factory',
+    'Web'
+)
+
+
+class SimpleLargeFileStorage(LargeFileStorage):
+
+    def __init__(self, repo: Path, endpoint: str):
+        self.repo = repo
+        self.endpoint = endpoint
+
+    def exists(self, oid: str) -> bool:
+        oid_path = self.path(oid)
+        return oid_path.exists()
+
+    def prepare_download(self, oid: str, size: int) -> BatchResponse.ObjectLfs.Action:
+        return self.prepare(oid)
+
+    def download(self, oid: str):
+        oid_path = self.path(oid)
+        return flask.helpers.send_file(str(oid_path))
+
+    def prepare_upload(self, oid: str, size: int) -> BatchResponse.ObjectLfs.Action:
+        return self.prepare(oid)
+
+    def upload(self, oid: str):
+        def mkdir(p):
+            try:
+                p.mkdir()
+            except FileExistsError:
+                pass
+
+        oid_path = self.path(oid)
+        mkdir(oid_path.parent.parent)
+        mkdir(oid_path.parent)
+
+        with open(oid_path, "wb") as f:
+            for chunk in wsgi.FileWrapper(request.stream):
+                f.write(chunk)
+
+    def prepare(self, oid: str) -> BatchResponse.ObjectLfs.Action:
+        return BatchResponse.ObjectLfs.Action(
+            href=f'{self.endpoint}transfer/{oid}'
+        )
+
+    def path(self, oid):
+        return self.repo / oid[:2] / oid[2:4] / oid
+
+
+class Factory:
+
+    @staticmethod
+    def create_large_file_storage():
+        return SimpleLargeFileStorage(
+            repo=Path(args.repo),
+            endpoint=args.endpoint if args.endpoint else request.url_root
+        )
+
+    @staticmethod
+    def create_batch_facade():
+        return BatchFacade(
+            Factory.create_large_file_storage()
+        )
+
+
+class Web:
+    app = flask.Flask(__name__)
+
+    @staticmethod
+    @app.route('/objects/batch', methods=['POST'])
+    def objects_batch():
+        facade = Factory.create_batch_facade()
+
+        response = facade.process(
+            HttpRequest(
+                path=request.path,
+                method=request.method,
+                headers=request.headers,
+                parameters=request.args,
+                body=json.dumps(request.json)
+            )
+        )
+
+        return dataclass_as_dict(response.body), response.status_code, response.headers
+
+    @staticmethod
+    @app.route('/transfer/<oid>', methods=['GET', 'PUT'])
+    def transfer(oid: str):
+        lfs = Factory.create_large_file_storage()
+
+        if request.method == 'GET':
+            if not lfs.exists(oid):
+                raise HttpError(404, 'Not found')
+
+            return lfs.download(oid), 200
+
+        elif request.method == 'PUT':
+            lfs.upload(oid)
+            return '', 202
+
+    @staticmethod
+    @app.errorhandler(404)
+    @app.errorhandler(405)
+    def not_found(e):
+        response = {
+            'message': 'Not found',
+            'request_id': str(uuid.uuid4())
+        }
+
+        return flask.jsonify(response), 404
+
+    @staticmethod
+    @app.errorhandler(HttpError)
+    def http_error(e):
+        response = {
+            'message': e.message,
+            'request_id': str(uuid.uuid4())
+        }
+
+        return flask.jsonify(response), e.code
+
+    @staticmethod
+    @app.errorhandler(Exception)
+    def internal_server_error(e):
+        Web.app.logger.error(e, exc_info=True)
+        response = {
+            'message': 'Internal Server Error',
+            'request_id': str(uuid.uuid4())
+        }
+
+        return flask.jsonify(response), 500
+
+
+if __name__ == '__main__':
+    parser = argparse.ArgumentParser(description='Git LFS: Self Hosted', add_help=False)
+    parser.add_argument('--port', type=int, default=5000, help="The port of the web server.")
+    parser.add_argument('--host', type=str, default='127.0.0.1', help="The hostname to listen on.")
+    parser.add_argument('--repo', type=str, required=True, help="Absolute path to Git LFS repository.")
+    parser.add_argument('--endpoint', type=str, required=False, default=None, help="Public endpoint address.")
+    parser.add_argument("--debug", action="store_true", help="Enable debug mode.")
+    parser.add_argument('-h', '--help', action='help', default=argparse.SUPPRESS, help='Print this message')
+    args = parser.parse_args()
+
+    web = Web()
+    web.app.run(port=args.port, host=args.host, debug=args.debug)

git-lfs-self/src/test/app_test.py

@@ -0,0 +1,172 @@
+import unittest
+from pathlib import Path
+from unittest import mock
+
+from app import *
+from core import *
+
+
+class SimpleLargeFileStorageTestCase(unittest.TestCase):
+
+    def setUp(self):
+        self.mock_repo = mock.MagicMock()
+        self.lfs = SimpleLargeFileStorage(
+            Path('foo'),
+            'https://example.com/'
+        )
+
+    def test_exists(self):
+        mock_path = mock.MagicMock()
+        mock_path.exists.return_value = True
+        self.lfs.path = mock.MagicMock()
+        self.lfs.path.return_value = mock_path
+
+        actual = self.lfs.exists('happyface.jpg')
+
+        self.assertIsNotNone(actual)
+        self.assertTrue(actual)
+        self.lfs.path.assert_called_once_with('happyface.jpg')
+        mock_path.exists.assert_called_once()
+
+    def test_not_exists(self):
+        mock_path = mock.MagicMock()
+        mock_path.exists.return_value = False
+        self.lfs.path = mock.MagicMock()
+        self.lfs.path.return_value = mock_path
+
+        actual = self.lfs.exists('happyface.jpg')
+
+        self.assertIsNotNone(actual)
+        self.assertFalse(actual)
+        self.lfs.path.assert_called_once_with('happyface.jpg')
+        mock_path.exists.assert_called_once()
+
+    def test_path(self):
+        actual = self.lfs.path('1bf0e3fc785fde')
+
+        self.assertIsNotNone(actual)
+        self.assertEqual(actual, Path('foo/1b/f0/1bf0e3fc785fde'))
+
+    def test_prepare(self):
+        actual = self.lfs.prepare('happyface.jpg')
+
+        self.assertIsNotNone(actual)
+        self.assertEqual(actual.href, 'https://example.com/transfer/happyface.jpg')
+
+    def test_prepare_download(self):
+        actual = self.lfs.prepare_download('happyface.jpg', 123)
+
+        self.assertIsNotNone(actual)
+        self.assertEqual(actual.href, 'https://example.com/transfer/happyface.jpg')
+
+    def test_prepare_upload(self):
+        actual = self.lfs.prepare_download('happyface.jpg', 123)
+
+        self.assertIsNotNone(actual)
+        self.assertEqual(actual.href, 'https://example.com/transfer/happyface.jpg')
+
+    def test_download(self):
+        self.lfs.path = mock.MagicMock()
+        self.lfs.path.return_value = 'foo'
+        with mock.patch('flask.helpers.send_file') as mock_send_file:
+            mock_send_file.return_value = 'baz'
+
+            actual = self.lfs.download('happyface.jpg')
+
+            self.assertEqual(actual, 'baz')
+            mock_send_file.assert_called_once_with('foo')
+            self.lfs.path.assert_called_once_with('happyface.jpg')
+
+    def test_upload(self):
+        # TODO: fix me
+        self.fail()
+
+
+class WebAppTestCase(unittest.TestCase):
+
+    def setUp(self):
+        Web.app.config['TESTING'] = True
+        Web.app.config['DEBUG'] = False
+        self.app = Web.app.test_client()
+
+    def tearDown(self):
+        pass
+
+    def test_not_found(self):
+        response = self.app.get('/foo')
+
+        self.assertEqual(response.status_code, 404)
+        self.assertEqual(response.json.get('message'), 'Not found')
+        self.assertIsNotNone(response.json.get('request_id'))
+
+    def test_internal_server_error(self):
+        with mock.patch('app.Factory.create_batch_facade') as mock_factory:
+            mock_facade = mock.MagicMock()
+            mock_factory.return_value = mock_facade
+            mock_facade.process.side_effect = ValueError()
+
+            response = self.app.post('/objects/batch')
+
+            self.assertEqual(response.status_code, 500)
+            self.assertEqual(response.json.get('message'), 'Internal Server Error')
+            self.assertIsNotNone(response.json.get('request_id'))
+
+    def test_http_error(self):
+        with mock.patch('app.Factory.create_batch_facade') as mock_factory:
+            mock_facade = mock.MagicMock()
+            mock_factory.return_value = mock_facade
+            mock_facade.process.side_effect = HttpError(402, 'foo')
+
+            response = self.app.post('/objects/batch')
+
+            self.assertEqual(response.status_code, 402)
+            self.assertEqual(response.json.get('message'), 'foo')
+            self.assertIsNotNone(response.json.get('request_id'))
+
+    def test_objects_batch(self):
+        with mock.patch('app.Factory.create_batch_facade') as mock_factory:
+            mock_facade = mock.MagicMock()
+            mock_facade.process.return_value = HttpResponse(BatchResponse('foo', []))
+            mock_factory.return_value = mock_facade
+
+            response = self.app.post('/objects/batch')
+
+            self.assertEqual(response.status_code, 200)
+            self.assertEqual(response.json, {'objects': [], 'transfer': 'foo'})
+
+    def test_transfer_get_not_found(self):
+        with mock.patch('app.Factory.create_large_file_storage') as mock_factory:
+            mock_lfs = mock.MagicMock()
+            mock_lfs.exists.return_value = False
+            mock_factory.return_value = mock_lfs
+
+            response = self.app.get('/transfer/123')
+
+            self.assertEqual(response.status_code, 404)
+            self.assertEqual(response.json.get('message'), 'Not found')
+            self.assertIsNotNone(response.json.get('request_id'))
+
+    def test_transfer_get(self):
+        with mock.patch('app.Factory.create_large_file_storage') as mock_factory:
+            mock_lfs = mock.MagicMock()
+            mock_lfs.exists.return_value = True
+            mock_lfs.download.return_value = 'foo'
+            mock_factory.return_value = mock_lfs
+
+            response = self.app.get('/transfer/123')
+
+            self.assertEqual(response.status_code, 200)
+            self.assertEqual(response.data, b'foo')
+            mock_lfs.exists.assert_called_once_with('123')
+            mock_lfs.download.assert_called_once_with('123')
+
+    def test_transfer_put(self):
+        with mock.patch('app.Factory.create_large_file_storage') as mock_factory:
+            mock_lfs = mock.MagicMock()
+            mock_factory.return_value = mock_lfs
+
+            response = self.app.put('/transfer/123')
+
+            self.assertEqual(response.status_code, 202)
+            self.assertEqual(response.data, b'')
+            mock_lfs.upload.assert_called_once_with('123')