ripgrep
til/python/stdlib-cli-tools.md
199 >>> import asyncio
200 >>> import httpx
201 >>> async with httpx.AsyncClient() as client:
202 ... r = await client.get('https://www.example.com/')
203 ...
til/pytest/registering-plugins-in-tests.md
53 @pytest.mark.asyncio
54 async def test_insert_alter(ds, unsafe):
55 async with httpx.AsyncClient(app=ds.app()) as client:
56 response = await client.post(
57 "http://localhost/-/insert/data/dogs?pk=id",
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
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
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
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
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-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-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-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-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-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-ripgrep/README.md
16 Some example searches:
17
18 - [with.\*AsyncClient](https://ripgrep.datasette.io/-/ripgrep?pattern=with.*AsyncClient) - regular expression search for `with.*AsyncClient`
19 - [.plugin_config, literal=on](https://ripgrep.datasette.io/-/ripgrep?pattern=.plugin_config\(&literal=on) - a non-regular expression search for `.plugin_config(`
20 - [with.\*AsyncClient glob=datasette/\*\*](https://ripgrep.datasette.io/-/ripgrep?pattern=with.*AsyncClient&glob=datasette%2F%2A%2A) - search for that pattern only within the `datasette/` top folder
21 - ["sqlite-utils\[">\] glob=setup.py](https://ripgrep.datasette.io/-/ripgrep?pattern=%22sqlite-utils%5B%22%3E%5D&glob=setup.py) - a regular expression search for packages that depend on either `sqlite-utils` or `sqlite-utils>=some-version`
22 - [test glob=!\*.html](https://ripgrep.datasette.io/-/ripgrep?pattern=test&glob=%21*.html) - search for the string `test` but exclude results in HTML files
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-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-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-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-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-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-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-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-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-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-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-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-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-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-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-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-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-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-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-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-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-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-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-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/docs/internals.rst
1286 async def fetch_url(url):
1287 with trace("fetch-url", url=url):
1288 async with httpx.AsyncClient() as client:
1289 return await client.get(url)
1290
datasette/datasette/app.py
1899
1900 async def get(self, path, **kwargs):
1901 async with httpx.AsyncClient(app=self.app) as client:
1902 return await client.get(self._fix(path), **kwargs)
1903
1904 async def options(self, path, **kwargs):
1905 async with httpx.AsyncClient(app=self.app) as client:
1906 return await client.options(self._fix(path), **kwargs)
1907
1908 async def head(self, path, **kwargs):
1909 async with httpx.AsyncClient(app=self.app) as client:
1910 return await client.head(self._fix(path), **kwargs)
1911
1912 async def post(self, path, **kwargs):
1913 async with httpx.AsyncClient(app=self.app) as client:
1914 return await client.post(self._fix(path), **kwargs)
1915
1916 async def put(self, path, **kwargs):
1917 async with httpx.AsyncClient(app=self.app) as client:
1918 return await client.put(self._fix(path), **kwargs)
1919
1920 async def patch(self, path, **kwargs):
1921 async with httpx.AsyncClient(app=self.app) as client:
1922 return await client.patch(self._fix(path), **kwargs)
1923
1924 async def delete(self, path, **kwargs):
1925 async with httpx.AsyncClient(app=self.app) as client:
1926 return await client.delete(self._fix(path), **kwargs)
1927
1928 async def request(self, method, path, **kwargs):
1929 avoid_path_rewrites = kwargs.pop("avoid_path_rewrites", None)
1930 async with httpx.AsyncClient(app=self.app) as client:
1931 return await client.request(
1932 method, self._fix(path, avoid_path_rewrites), **kwargs