ripgrep
datasette/datasette/app.py
1754
1755 async def get(self, path, **kwargs):
1756 async with httpx.AsyncClient(app=self.app) as client:
1757 return await client.get(self._fix(path), **kwargs)
1758
1759 async def options(self, path, **kwargs):
1760 async with httpx.AsyncClient(app=self.app) as client:
1761 return await client.options(self._fix(path), **kwargs)
1762
1763 async def head(self, path, **kwargs):
1764 async with httpx.AsyncClient(app=self.app) as client:
1765 return await client.head(self._fix(path), **kwargs)
1766
1767 async def post(self, path, **kwargs):
1768 async with httpx.AsyncClient(app=self.app) as client:
1769 return await client.post(self._fix(path), **kwargs)
1770
1771 async def put(self, path, **kwargs):
1772 async with httpx.AsyncClient(app=self.app) as client:
1773 return await client.put(self._fix(path), **kwargs)
1774
1775 async def patch(self, path, **kwargs):
1776 async with httpx.AsyncClient(app=self.app) as client:
1777 return await client.patch(self._fix(path), **kwargs)
1778
1779 async def delete(self, path, **kwargs):
1780 async with httpx.AsyncClient(app=self.app) as client:
1781 return await client.delete(self._fix(path), **kwargs)
1782
1783 async def request(self, method, path, **kwargs):
1784 avoid_path_rewrites = kwargs.pop("avoid_path_rewrites", None)
1785 async with httpx.AsyncClient(app=self.app) as client:
1786 return await client.request(
1787 method, self._fix(path, avoid_path_rewrites), **kwargs
datasette/docs/internals.rst
1224 async def fetch_url(url):
1225 with trace("fetch-url", url=url):
1226 async with httpx.AsyncClient() as client:
1227 return await client.get(url)
1228
datasette-allow-permissions-debug/tests/test_allow_permissions_debug.py
7 async def test_plugin_is_installed():
8 app = Datasette([], memory=True).app()
9 async with httpx.AsyncClient(app=app) as client:
10 response = await client.get("http://localhost/-/plugins.json")
11 assert 200 == response.status_code
17 async def test_allows_access():
18 app = Datasette([], memory=True).app()
19 async with httpx.AsyncClient(app=app) as client:
20 response = await client.get("http://localhost/-/permissions")
21 assert 200 == response.status_code
datasette-auth-existing-cookies/datasette_auth_existing_cookies/__init__.py
47 return datasette._auth_existing_cookies_cache[cache_key]
48
49 async with httpx.AsyncClient() as client:
50 response = await client.get(api_url, params=header_params, cookies=cookies)
51
datasette-auth-github/datasette_auth_github/utils.py
13 org, profile["login"]
14 )
15 async with httpx.AsyncClient() as client:
16 response = await client.get(
17 url, headers={"Authorization": "token {}".format(access_token)}
30 org_slug, team_slug
31 )
32 async with httpx.AsyncClient() as client:
33 response = await client.get(
34 lookup_url,
45 )
46 )
47 async with httpx.AsyncClient() as client:
48 response = await client.get(
49 team_membership_url,
datasette-auth-github/datasette_auth_github/views.py
45
46 # Exchange that code for a token
47 async with httpx.AsyncClient() as client:
48 github_response = await client.post(
49 "https://github.com/login/oauth/access_token",
67 profile_url = "https://api.github.com/user"
68 try:
69 async with httpx.AsyncClient() as client:
70 profile = (
71 await client.get(
datasette-auth-tokens/tests/test_auth_tokens.py
67 @pytest.mark.asyncio
68 async def test_token(ds, token, path, expected_status):
69 async with httpx.AsyncClient(app=ds.app()) as client:
70 response = await client.get(
71 "http://localhost{}".format(path),
88 @pytest.mark.asyncio
89 async def test_query_param(ds, token, path, expected_status):
90 async with httpx.AsyncClient(app=ds.app()) as client:
91 response = await client.get(
92 "http://localhost{}&_auth_token={}".format(path, token),
110 @pytest.mark.asyncio
111 async def test_query(ds, token, path, expected_status):
112 async with httpx.AsyncClient(app=ds.app()) as client:
113 response = await client.get(
114 "http://localhost{}".format(path),
129 @pytest.mark.asyncio
130 async def test_actor(ds, token, expected_actor):
131 async with httpx.AsyncClient(app=ds.app()) as client:
132 response = await client.get(
133 "http://localhost/-/actor.json",
142 @pytest.mark.asyncio
143 async def test_tokens_table_not_visible(ds, path):
144 async with httpx.AsyncClient(app=ds.app()) as client:
145 response = await client.get("http://localhost{}".format(path))
146 assert 403 == response.status_code
datasette-backup/tests/test_backup.py
22 async def test_plugin_is_installed():
23 app = Datasette([], memory=True).app()
24 async with httpx.AsyncClient(app=app) as client:
25 response = await client.get("http://localhost/-/plugins.json")
26 assert 200 == response.status_code
31 @pytest.mark.asyncio
32 async def test_backup_sql(ds):
33 async with httpx.AsyncClient(app=ds.app()) as client:
34 assert (
35 await client.get("http://localhost/-/backup/nope.sql")
65 db["dogs"].enable_fts(["name"])
66 ds = Datasette([db_path])
67 async with httpx.AsyncClient(app=ds.app()) as client:
68 response = await client.get("http://localhost/-/backup/fts.sql")
69 assert response.status_code == 200
datasette-big-local/datasette_big_local/__init__.py
123
124 async def get_project(datasette, project_id, remember_token, files=False):
125 async with httpx.AsyncClient() as client:
126 response = await client.post(
127 get_settings(datasette).graphql_url,
211 """,
212 }
213 async with httpx.AsyncClient() as client:
214 response = await client.post(
215 graphql_endpoint,
362 }
363 """.strip()
364 async with httpx.AsyncClient() as client:
365 response = await client.post(
366 graphql_endpoint,
datasette-block/tests/test_block.py
16 app = datasette.app()
17 async with LifespanManager(app):
18 async with httpx.AsyncClient(app=app) as client:
19 response = await client.get("http://localhost" + path)
20 assert response.status_code == expected_status_code
datasette-column-inspect/tests/test_column_inspect.py
22 async def test_table_page_has_script_on_it(db_path):
23 app = Datasette([db_path]).app()
24 async with httpx.AsyncClient(app=app) as client:
25 response = await client.get("http://localhost/data/creatures")
26 assert 200 == response.status_code
datasette-dateutil/tests/test_dateutil.py
7 async def test_plugin_is_installed():
8 app = Datasette([], memory=True).app()
9 async with httpx.AsyncClient(app=app) as client:
10 response = await client.get("http://localhost/-/plugins.json")
11 assert 200 == response.status_code
85 async def test_dateutil_sql_functions(sql, expected):
86 app = Datasette([], memory=True).app()
87 async with httpx.AsyncClient(app=app) as client:
88 response = await client.get(
89 "http://localhost/_memory.json",
101 async def test_dateutil_unbounded_rrule_error():
102 app = Datasette([], memory=True).app()
103 async with httpx.AsyncClient(app=app) as client:
104 response = await client.get(
105 "http://localhost/_memory.json",
datasette-dns/tests/test_dns.py
10 async def test_plugin_is_installed():
11 app = Datasette([], memory=True).app()
12 async with httpx.AsyncClient(app=app) as client:
13 response = await client.get("http://localhost/-/plugins.json")
14 assert 200 == response.status_code
22 m.side_effect = dns.resolver.NoAnswer
23 app = Datasette([], memory=True).app()
24 async with httpx.AsyncClient(app=app) as client:
25 response = await client.get(
26 "http://localhost/:memory:.json?"
datasette-glitch/tests/test_glitch.py
8 async def test_plugin_is_installed():
9 app = Datasette([], memory=True).app()
10 async with httpx.AsyncClient(app=app) as client:
11 response = await client.get("http://localhost/-/plugins.json")
12 assert 200 == response.status_code
datasette-import-table/datasette_import_table/__init__.py
107
108 while url:
109 async with httpx.AsyncClient() as client:
110 response = await client.get(url)
111 data = response.json()
146 async def load_first_page(url):
147 url = url + ".json?_shape=objects&_size=max"
148 async with httpx.AsyncClient() as client:
149 response = await client.get(url)
150 if response.status_code != 200:
datasette-indieauth/datasette_indieauth/__init__.py
125 "code_verifier": code_verifier,
126 }
127 async with httpx.AsyncClient() as client:
128 response = await client.post(authorization_endpoint, data=data)
129
datasette-indieauth/datasette_indieauth/utils.py
131 canonical_url = None
132 chunk = None
133 async with httpx.AsyncClient(max_redirects=5) as client:
134 try:
135 response = await client.get(url, follow_redirects=True)
datasette-init/tests/test_init.py
35 )
36 await ds.invoke_startup()
37 async with httpx.AsyncClient(app=ds.app()) as client:
38 response = await client.get("http://localhost/test/dogs.json")
39 assert 200 == response.status_code
67 )
68 await ds.invoke_startup()
69 async with httpx.AsyncClient(app=ds.app()) as client:
70 response = await client.get("http://localhost/test/dogs.json")
71 assert 200 == response.status_code
79 ds = build_datasette(tmp_path_factory, {"views": {"two": "select 1 + 1"}})
80 await ds.invoke_startup()
81 async with httpx.AsyncClient(app=ds.app()) as client:
82 response = await client.get("http://localhost/test/two.json?_shape=array")
83 assert 200 == response.status_code
92 db_init=lambda db: db.create_view("two", "select 1 + 3"),
93 )
94 async with httpx.AsyncClient(app=ds.app()) as client:
95 response = await client.get("http://localhost/test/two.json?_shape=array")
96 assert [{"1 + 3": 4}] == response.json()
104 ds = Datasette([], memory=True)
105 await ds.invoke_startup()
106 async with httpx.AsyncClient(app=ds.app()) as client:
107 assert 200 == (await client.get("http://localhost/")).status_code
datasette-insert-unsafe/tests/test_insert_unsafe.py
18 async def test_plugin_is_installed():
19 app = Datasette([], memory=True).app()
20 async with httpx.AsyncClient(app=app) as client:
21 response = await client.get("http://localhost/-/plugins.json")
22 assert 200 == response.status_code
28 async def test_insert_allowed(db_path):
29 app = Datasette([str(db_path)]).app()
30 async with httpx.AsyncClient(app=app) as client:
31 response = await client.post(
32 "http://localhost/-/insert/data/newtable", json=[{"foo": "bar"}],
40 plugin = pm.unregister(name="insert_unsafe")
41 try:
42 async with httpx.AsyncClient(app=app) as client:
43 response = await client.post(
44 "http://localhost/-/insert/data/newtable", json=[{"foo": "bar"}],
datasette-mask-columns/tests/test_mask_columns.py
12 datasette = Datasette([path], memory=True)
13 # Without the plugin:
14 async with httpx.AsyncClient(app=datasette.app()) as client:
15 response = await client.get("http://localhost/foo/users.json?_shape=array")
16 assert 200 == response.status_code
29 },
30 )
31 async with httpx.AsyncClient(app=datasette2.app()) as client:
32 response = await client.get("http://localhost/foo/users.json?_shape=array")
33 assert 200 == response.status_code
datasette-media/datasette_media/__init__.py
75 if should_transform:
76 if content is None and content_url:
77 async with httpx.AsyncClient() as client:
78 response = await client.get(row["content_url"])
79 content = response.content
datasette-media/tests/test_media.py
24 },
25 ).app()
26 async with httpx.AsyncClient(app=app) as client:
27 response = await client.get("http://localhost/-/media/photos/key")
28 assert 200 == response.status_code
51 },
52 ).app()
53 async with httpx.AsyncClient(app=app) as client:
54 response = await client.get("http://localhost/-/media/text/key")
55 assert 200 == response.status_code
153 },
154 ).app()
155 async with httpx.AsyncClient(app=app) as client:
156 response = await client.get("http://localhost/-/media/photos/1")
157 assert 200 == response.status_code
183 },
184 ).app()
185 async with httpx.AsyncClient(app=app) as client:
186 response = await client.get("http://localhost/-/media/photos/1")
187 assert 200 == response.status_code
210 },
211 ).app()
212 async with httpx.AsyncClient(app=app) as client:
213 response = await client.get("http://localhost/-/media/photos/1")
214 assert 200 == response.status_code
239 },
240 ).app()
241 async with httpx.AsyncClient(app=app) as client:
242 response = await client.get("http://localhost/-/media/photos/1")
243 assert 200 == response.status_code
279 },
280 ).app()
281 async with httpx.AsyncClient(app=app) as client:
282 response = await client.get("http://localhost/-/media/photos/1", params=args)
283 assert 200 == response.status_code
datasette-permissions-sql/tests/test_permissions_sql.py
90 @pytest.mark.asyncio
91 async def test_permissions_sql(ds, actor, table, expected_status):
92 async with httpx.AsyncClient(app=ds.app()) as client:
93 cookies = {}
94 if actor:
105 @pytest.mark.asyncio
106 async def test_fallback(ds, actor, expected_status):
107 async with httpx.AsyncClient(app=ds.app()) as client:
108 cookies = {}
109 if actor:
datasette-plugin-demos/tests/test_plugin_demos.py
7 async def test_plugin_is_installed():
8 app = Datasette([], memory=True).app()
9 async with httpx.AsyncClient(app=app) as client:
10 response = await client.get("http://localhost/-/plugins.json")
11 assert 200 == response.status_code
datasette-psutil/tests/test_psutil.py
7 async def test_datasette_psutil():
8 ds = Datasette([], memory=True)
9 async with httpx.AsyncClient(app=ds.app()) as client:
10 response = await client.get("http://localhost/-/psutil")
11 assert 200 == response.status_code
datasette-redirect-to-https/tests/test_redirect_to_https.py
21 datasette = Datasette([], memory=True)
22 async with LifespanManager(datasette.app()):
23 async with httpx.AsyncClient(app=datasette.app()) as client:
24 response = await client.get("https://localhost{}".format(path))
25 assert response.status_code == 200
datasette-remote-metadata/datasette_remote_metadata/__init__.py
24 fetch_url += "&" if "?" in url else "?"
25 fetch_url += str(random.random())
26 async with httpx.AsyncClient() as client:
27 response = await client.get(
28 fetch_url,
datasette-saved-queries/tests/test_saved_queries.py
22 @pytest.mark.asyncio
23 async def test_plugin_is_installed(ds):
24 async with httpx.AsyncClient(app=ds.app()) as client:
25 response = await client.get("http://localhost/-/plugins.json")
26 assert 200 == response.status_code
31 @pytest.mark.asyncio
32 async def test_table_created(ds):
33 async with httpx.AsyncClient(app=ds.app()) as client:
34 response = await client.get("http://localhost/data/saved_queries.json")
35 assert 200 == response.status_code
39 @pytest.mark.asyncio
40 async def test_save_query(ds):
41 async with httpx.AsyncClient(app=ds.app()) as client:
42 # Get the csrftoken cookie
43 response1 = await client.get("http://localhost/data/save_query")
69 @pytest.mark.asyncio
70 async def test_save_query_authenticated_actor(ds):
71 async with httpx.AsyncClient(app=ds.app()) as client:
72 response1 = await client.get("http://localhost/data/save_query")
73 response2 = await client.post(
datasette-seaborn/tests/test_seaborn.py
24 async def test_requires_seaborn(ds):
25 app = ds.app()
26 async with httpx.AsyncClient(app=app) as client:
27 response = await client.get("http://localhost/penguins/penguins.seaborn")
28 assert response.status_code == 500
36
37 async def get_dims(path):
38 async with httpx.AsyncClient(app=app) as client:
39 response = await client.get("http://localhost{}".format(path))
40 assert response.status_code == 200
datasette-socrata/datasette_socrata/__init__.py
56 metadata_url = "https://{}/api/views/{}.json".format(domain, id)
57 try:
58 async with httpx.AsyncClient() as client:
59 metadata_response = await client.get(metadata_url)
60 if metadata_response.status_code != 200:
68 async def get_row_count(domain, id):
69 # Fetch the row count too - we ignore errors and keep row_count at None
70 async with httpx.AsyncClient() as client:
71 count_url = "https://{}/resource/{}.json?$select=count(*)".format(domain, id)
72 count_response = await client.get(count_url)
259 return await database.execute_write_fn(_write, block=True)
260
261 async with httpx.AsyncClient() as client:
262 async with client.stream("GET", csv_url) as response:
263 reader = AsyncDictReader(response.aiter_lines())
datasette-template-sql/tests/test_template_sql.py
28 @pytest.mark.asyncio
29 async def test_sql_against_named_database(app):
30 async with httpx.AsyncClient(app=app) as client:
31 response = await client.get("http://localhost/news")
32 assert 200 == response.status_code
40 @pytest.mark.asyncio
41 async def test_sql_against_default_database(app):
42 async with httpx.AsyncClient(app=app) as client:
43 response = await client.get("http://localhost/")
44 assert 200 == response.status_code
49 @pytest.mark.asyncio
50 async def test_sql_with_arguments(app):
51 async with httpx.AsyncClient(app=app) as client:
52 response = await client.get("http://localhost/news")
53 assert 200 == response.status_code
datasette-upload-csvs/tests/test_datasette_upload_csvs.py
15 app = ds.app()
16 async with LifespanManager(app):
17 async with httpx.AsyncClient(app=app) as client:
18 response = await client.get("http://localhost/")
19 assert 200 == response.status_code
23 async def test_redirect():
24 datasette = Datasette([], memory=True)
25 async with httpx.AsyncClient(app=datasette.app()) as client:
26 response = await client.get("http://localhost/-/upload-csv")
27 assert response.status_code == 302
39 app = ds.app()
40 async with LifespanManager(app):
41 async with httpx.AsyncClient(app=app) as client:
42 cookies = {}
43 if auth:
137
138 # First test the upload page exists
139 async with httpx.AsyncClient(app=datasette.app()) as client:
140 response = await client.get("http://localhost/-/upload-csvs", cookies=cookies)
141 assert 200 == response.status_code
193 ds = Datasette([path])
194 app = ds.app()
195 async with httpx.AsyncClient(app=app) as client:
196 response = await client.get("http://localhost/-/upload-csvs")
197 assert 403 == response.status_code
198 # Now try with a root actor
199 async with httpx.AsyncClient(app=app) as client2:
200 response2 = await client2.get(
201 "http://localhost/-/upload-csvs",
datasette-vega/tests/test_vega.py
7 async def test_plugin_is_installed():
8 app = Datasette([], memory=True).app()
9 async with httpx.AsyncClient(app=app) as client:
10 response = await client.get("http://localhost/-/plugins.json")
11 assert response.status_code == 200
datasette-write/tests/test_write.py
85 @pytest.mark.asyncio
86 async def test_execute_write(ds, database, sql, expected_message):
87 async with httpx.AsyncClient(
88 app=ds.app(), cookies={"ds_actor": ds.sign({"a": {"id": "root"}}, "actor")}
89 ) as client:
datasette-yaml/tests/test_yaml.py
9 async def test_plugin_is_installed():
10 app = Datasette([], memory=True).app()
11 async with httpx.AsyncClient(app=app) as client:
12 response = await client.get("http://localhost/-/plugins.json")
13 assert 200 == response.status_code
29 )
30 app = Datasette([str(db_path)]).app()
31 async with httpx.AsyncClient(app=app) as client:
32 response = await client.get("http://localhost/test/dogs.yaml")
33 assert response.status_code == 200
dogsheep-beta/tests/test_plugin.py
12 @pytest.mark.asyncio
13 async def test_search(ds):
14 async with httpx.AsyncClient(app=ds.app()) as client:
15 response = await client.get("http://localhost/-/beta")
16 assert 200 == response.status_code
141 )
142 async def test_advanced_search(ds, q, expected):
143 async with httpx.AsyncClient(app=ds.app()) as client:
144 response = await client.get(
145 "http://localhost/-/beta?" + urllib.parse.urlencode({"q": q})
173 )
174 async def test_search_order(ds, sort, expected):
175 async with httpx.AsyncClient(app=ds.app()) as client:
176 q = "email"
177 response = await client.get(
213 )
214 async def test_search_order_for_timeline(ds, sort, expected):
215 async with httpx.AsyncClient(app=ds.app()) as client:
216 url = "http://localhost/-/beta"
217 if sort:
234 @pytest.mark.asyncio
235 async def test_fixture(ds):
236 async with httpx.AsyncClient(app=ds.app()) as client:
237 response = await client.get("http://localhost/-/databases.json")
238 assert 200 == response.status_code
243 async def test_plugin_is_installed():
244 app = Datasette([], memory=True).app()
245 async with httpx.AsyncClient(app=app) as client:
246 response = await client.get("http://localhost/-/plugins.json")
247 assert 200 == response.status_code
til/asgi/lifespan-test-httpx.md
23 app = ds.app()
24 async with LifespanManager(app):
25 async with httpx.AsyncClient(app=app) as client:
26 response = await client.get("http://localhost/-/asgi-scope")
27 assert 200 == response.status_code
til/git/git-bisect.md
29 async def run_check():
30 ds = Datasette([])
31 async with httpx.AsyncClient(app=ds.app()) as client:
32 response = await client.get("http://localhost/")
33 assert 200 == response.status_code